diff --git a/README.md b/README.md index 318ab10..0f9e0fc 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ v0.2.0 当前正式发布目标为 **macOS**,GitHub Release 同时生成: 1. **启用数据源**:打开「设置 → 数据源」,确认本机已安装的 AI 工具与扫描目录。启动扫描只检测变化并写入本地 SQLite,不会调用模型。 2. **同步会话**:点击顶栏「同步」,再从「进度中心」查看扫描、解析、写入进度和失败文件;取消会在当前文件处理完成后生效。 -3. **配置分析 Provider**:打开「设置 → AI 配置」,新增 API 或 CLI 配置并设为当前。API 支持刷新模型列表或手填完整模型 ID;CLI 的「检测命令」只检查可执行性,不调用模型。 +3. **配置分析 Provider**:打开「设置 → AI 配置」,新增 API 或 CLI 配置并设为当前,同时按本机和 Provider 承载能力设置 1–8 个全局并行分析。API 支持刷新模型列表或手填完整模型 ID;CLI 的「检测命令」只检查可执行性,不调用模型。 4. **确认并开始分析**:在「对话档案」预览会话,选择单条或批量分析。运行期间不会自动跨 Provider 切换;失败后可为单项选择其他配置重试。 5. **审核与复用知识**:在「知识笔记」中编辑类型、摘要、正文、主题标签和技术项。可全文搜索、合并标签、发布草稿,或将选中/全部知识导出为 Markdown。 diff --git a/apps/desktop/src-tauri/src/commands/config.rs b/apps/desktop/src-tauri/src/commands/config.rs index a1c5393..a3f352f 100644 --- a/apps/desktop/src-tauri/src/commands/config.rs +++ b/apps/desktop/src-tauri/src/commands/config.rs @@ -21,6 +21,7 @@ pub struct AppConfigDto { #[serde(rename_all = "camelCase")] pub struct DistillerConfigDto { pub active_profile_id: String, + pub max_concurrent_analyses: usize, pub profiles: Vec, } @@ -64,6 +65,7 @@ impl From<&AppConfig> for AppConfigDto { Self { distiller: DistillerConfigDto { active_profile_id: c.distiller.active_profile_id.clone(), + max_concurrent_analyses: c.distiller.max_concurrent_analyses, profiles: c .distiller .profiles @@ -128,6 +130,7 @@ impl From for AppConfig { Self { distiller: DistillerConfig { active_profile_id: dto.distiller.active_profile_id, + max_concurrent_analyses: dto.distiller.max_concurrent_analyses, profiles: dto .distiller .profiles @@ -198,6 +201,9 @@ pub async fn save_config(state: State<'_, AppState>, config: AppConfigDto) -> Re new_config .save(&AppConfig::default_path()) .map_err(|e| format!("配置保存失败: {e}"))?; + state + .analysis_runtime + .set_limit(new_config.distiller.max_concurrent_analyses); *state .config .write() diff --git a/apps/desktop/src-tauri/src/commands/jobs.rs b/apps/desktop/src-tauri/src/commands/jobs.rs index debd9cb..b932ded 100644 --- a/apps/desktop/src-tauri/src/commands/jobs.rs +++ b/apps/desktop/src-tauri/src/commands/jobs.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use tauri::{AppHandle, Emitter, State}; @@ -8,7 +8,7 @@ use crate::config::AppConfig; use crate::sidecar::SidecarManager; use crate::storage::models::{Job, NewJobItem}; use crate::storage::Database; -use crate::AppState; +use crate::{AnalysisRuntime, AppState}; fn emit_job(app: &AppHandle, db: &Database, job_id: &str) { if let Ok(job) = db.get_job_snapshot(job_id) { @@ -21,163 +21,270 @@ fn spawn_job( db: Arc, config: AppConfig, sidecar: Option>, - job_lock: Arc>, + job_lock: Arc>, + analysis_runtime: Arc, job_id: String, ) { tauri::async_runtime::spawn(async move { - let _guard = job_lock.lock().await; let Ok(job) = db.get_job(&job_id) else { return }; - if job.cancel_requested { - let _ = db.cancel_queued_items(&job_id); - let _ = db.finish_job(&job_id, "cancelled", None); - emit_job(&app, &db, &job_id); - return; - } - let phase = if job.kind == "sync" { - "scanning" + if job.kind == "sync" { + let _guard = job_lock.write().await; + run_sync_job(app, db, config, job).await; } else { - "judging" - }; - if db.mark_job_running(&job_id, phase).is_err() { - return; + let _guard = job_lock.read().await; + run_analysis_job(app, db, config, sidecar, analysis_runtime, job).await; } + }); +} + +async fn run_sync_job(app: AppHandle, db: Arc, config: AppConfig, job: Job) { + let job_id = job.id.clone(); + if job.cancel_requested { + let _ = db.cancel_queued_items(&job_id); + let _ = db.finish_job(&job_id, "cancelled", None); emit_job(&app, &db, &job_id); + return; + } + if db.mark_job_running(&job_id, "scanning").is_err() { + return; + } + emit_job(&app, &db, &job_id); - // 一个分析任务只初始化一次供应商;任务内所有条目共享同一 Sidecar 与配置快照。 - if job.kind == "analysis" { - let init_result = sidecar - .as_ref() - .ok_or_else(|| "Sidecar 未就绪".to_string()) - .and_then(|sidecar| { - let params = config.sidecar_init_params(job.provider_profile_id.as_deref())?; - sidecar - .call_with_timeout::( - "init", - params, - std::time::Duration::from_secs(30), - ) - .map_err(|error| format!("Sidecar init 失败:{error}")) + let mut failed = false; + for item in job.items { + if db.job_cancel_requested(&job_id).unwrap_or(false) { + break; + } + if db.mark_item_running(&item.id, "scanning").is_err() { + failed = true; + continue; + } + emit_job(&app, &db, &job_id); + let started = Instant::now(); + let source_id = item.source_id.clone().unwrap_or_default(); + let progress_db = db.clone(); + let progress_app = app.clone(); + let progress_job_id = job_id.clone(); + let progress_source_id = source_id.clone(); + let worker_db = db.clone(); + let worker_config = config.clone(); + let result = tokio::task::spawn_blocking(move || { + let result = CollectorScheduler::new(&worker_config, &worker_db) + .collect_source_with_progress(&source_id, |raw_path, phase, error| { + if let Ok(id) = progress_db.append_job_item( + &progress_job_id, + &NewJobItem { + session_id: None, + source_id: Some(&progress_source_id), + raw_path: Some(raw_path), + }, + ) { + let status = if error.is_some() { + "failed" + } else { + "succeeded" + }; + let _ = + progress_db.finish_item(&progress_job_id, &id, status, phase, 0, error); + emit_job(&progress_app, &progress_db, &progress_job_id); + } }); - if let Err(error) = init_result { - for item in &job.items { - let _ = db.finish_item(&job_id, &item.id, "failed", "failed", 0, Some(&error)); - } - let _ = db.finish_job(&job_id, "failed", Some(&error)); - emit_job(&app, &db, &job_id); - return; + if result.failed > 0 { + Err(format!("{} 个文件同步失败", result.failed)) + } else { + Ok(()) } - } + }) + .await + .map_err(|error| error.to_string()) + .and_then(|value| value); - let mut failed = false; - for item in job.items { - if db.job_cancel_requested(&job_id).unwrap_or(false) { - break; + let duration = started.elapsed().as_millis().min(i64::MAX as u128) as i64; + let cancelled = db.job_cancel_requested(&job_id).unwrap_or(false); + match (result, cancelled) { + (_, true) => { + let _ = db.finish_item( + &job_id, + &item.id, + "cancelled", + "cancelled", + duration, + Some("用户取消"), + ); } - let item_phase = if job.kind == "sync" { - "scanning" - } else { - "judging" - }; - if db.mark_item_running(&item.id, item_phase).is_err() { + (Ok(()), false) => { + let _ = db.finish_item(&job_id, &item.id, "succeeded", "completed", duration, None); + } + (Err(error), false) => { failed = true; - continue; + let _ = db.finish_item( + &job_id, + &item.id, + "failed", + "failed", + duration, + Some(&error), + ); } - emit_job(&app, &db, &job_id); - let started = Instant::now(); + } + emit_job(&app, &db, &job_id); + } - let result: Result<(), String> = if job.kind == "sync" { - let db = db.clone(); - let config = config.clone(); - let source_id = item.source_id.clone().unwrap_or_default(); - let progress_db = db.clone(); - let progress_app = app.clone(); - let progress_job_id = job_id.clone(); - let progress_source_id = source_id.clone(); - tokio::task::spawn_blocking(move || { - let result = CollectorScheduler::new(&config, &db) - .collect_source_with_progress(&source_id, |raw_path, phase, error| { - if let Ok(id) = progress_db.append_job_item( - &progress_job_id, - &NewJobItem { - session_id: None, - source_id: Some(&progress_source_id), - raw_path: Some(raw_path), - }, - ) { - let status = if error.is_some() { - "failed" - } else { - "succeeded" - }; - let _ = progress_db.finish_item( - &progress_job_id, - &id, - status, - phase, - 0, - error, - ); - emit_job(&progress_app, &progress_db, &progress_job_id); - } - }); - if result.failed > 0 { - Err(format!("{} 个文件同步失败", result.failed)) - } else { - Ok(()) - } - }) - .await - .map_err(|error| error.to_string()) - .and_then(|value| value) - } else { - let db = db.clone(); - let config = config.clone(); - let sidecar = sidecar.clone().ok_or_else(|| "Sidecar 未就绪".to_string()); - let session_id = item - .session_id - .clone() - .ok_or_else(|| "任务条目缺少会话".to_string()); - let profile_id = job.provider_profile_id.clone(); - match (sidecar, session_id) { - (Ok(sidecar), Ok(session_id)) => { - let phase_db = db.clone(); - let phase_app = app.clone(); - let phase_job_id = job_id.clone(); - let phase_item_id = item.id.clone(); - tokio::task::spawn_blocking(move || { - let on_phase = |phase: &str| { - let _ = phase_db.update_job_item_phase( - &phase_job_id, - &phase_item_id, - phase, - ); - emit_job(&phase_app, &phase_db, &phase_job_id); - }; - super::sessions::run_distill_pipeline( - &db, - &config, - &sidecar, - &session_id, - &phase_job_id, - profile_id.as_deref(), - Some(&on_phase), - ) - .map(|_| ()) - }) - .await - .map_err(|error| error.to_string()) - .and_then(|value| value) - } - (Err(error), _) | (_, Err(error)) => Err(error), + finish_job(&app, &db, &job_id, failed); +} + +async fn run_analysis_job( + app: AppHandle, + db: Arc, + config: AppConfig, + sidecar: Option>, + analysis_runtime: Arc, + job: Job, +) { + let job_id = job.id.clone(); + if job.cancel_requested { + let _ = db.cancel_queued_items(&job_id); + let _ = db.finish_job(&job_id, "cancelled", None); + emit_job(&app, &db, &job_id); + return; + } + if db.mark_job_running(&job_id, "judging").is_err() { + return; + } + emit_job(&app, &db, &job_id); + + let Some(base_sidecar) = sidecar else { + fail_all_analysis_items(&app, &db, &job, "Sidecar 未就绪"); + return; + }; + let init_params = match config.sidecar_init_params(job.provider_profile_id.as_deref()) { + Ok(params) => params, + Err(error) => { + fail_all_analysis_items(&app, &db, &job, &error); + return; + } + }; + + let mut tasks = tokio::task::JoinSet::new(); + for item in job.items { + let task_app = app.clone(); + let task_db = db.clone(); + let task_config = config.clone(); + let task_sidecar = base_sidecar.clone(); + let task_init_params = init_params.clone(); + let task_runtime = analysis_runtime.clone(); + let task_job_id = job_id.clone(); + let profile_id = job.provider_profile_id.clone(); + tasks.spawn(async move { + let _permit = loop { + if task_db.job_cancel_requested(&task_job_id).unwrap_or(false) { + return false; + } + if let Ok(permit) = + tokio::time::timeout(Duration::from_millis(100), task_runtime.acquire()).await + { + break permit; + } + }; + if task_db.job_cancel_requested(&task_job_id).unwrap_or(false) { + return false; + } + if task_db.mark_item_running(&item.id, "judging").is_err() { + return true; + } + emit_job(&task_app, &task_db, &task_job_id); + let started = Instant::now(); + let session_id = match item.session_id.clone() { + Some(session_id) => session_id, + None => { + let _ = task_db.finish_item( + &task_job_id, + &item.id, + "failed", + "failed", + 0, + Some("任务条目缺少会话"), + ); + emit_job(&task_app, &task_db, &task_job_id); + return true; } }; + let task_worker = Arc::new(task_sidecar.worker()); + task_runtime.register_worker(&task_job_id, &item.id, task_worker.clone()); + if task_db.job_cancel_requested(&task_job_id).unwrap_or(false) { + let _ = task_worker.cancel(); + let _ = task_db.finish_item( + &task_job_id, + &item.id, + "cancelled", + "cancelled", + started.elapsed().as_millis().min(i64::MAX as u128) as i64, + Some("用户取消"), + ); + task_runtime.unregister_worker(&task_job_id, &item.id); + emit_job(&task_app, &task_db, &task_job_id); + return false; + } + let init_worker = task_worker.clone(); + let init_result = tokio::task::spawn_blocking(move || { + init_worker + .call_with_timeout::( + "init", + task_init_params, + Duration::from_secs(30), + ) + .map_err(|error| format!("Sidecar init 失败:{error}")) + }) + .await + .map_err(|error| error.to_string()) + .and_then(|value| value); + if let Err(error) = init_result { + let cancelled = task_db.job_cancel_requested(&task_job_id).unwrap_or(false); + let _ = task_db.finish_item( + &task_job_id, + &item.id, + if cancelled { "cancelled" } else { "failed" }, + if cancelled { "cancelled" } else { "failed" }, + started.elapsed().as_millis().min(i64::MAX as u128) as i64, + Some(if cancelled { "用户取消" } else { &error }), + ); + task_runtime.unregister_worker(&task_job_id, &item.id); + emit_job(&task_app, &task_db, &task_job_id); + return !cancelled; + } + let phase_db = task_db.clone(); + let phase_app = task_app.clone(); + let phase_job_id = task_job_id.clone(); + let phase_item_id = item.id.clone(); + let pipeline_db = task_db.clone(); + let pipeline_job_id = task_job_id.clone(); + let result = tokio::task::spawn_blocking(move || { + let on_phase = |phase: &str| { + let _ = phase_db.update_job_item_phase(&phase_job_id, &phase_item_id, phase); + emit_job(&phase_app, &phase_db, &phase_job_id); + }; + super::sessions::run_distill_pipeline( + &pipeline_db, + &task_config, + &task_worker, + &session_id, + &pipeline_job_id, + profile_id.as_deref(), + Some(&on_phase), + ) + .map(|_| ()) + }) + .await + .map_err(|error| error.to_string()) + .and_then(|value| value); let duration = started.elapsed().as_millis().min(i64::MAX as u128) as i64; - let cancelled = db.job_cancel_requested(&job_id).unwrap_or(false); + let cancelled = task_db.job_cancel_requested(&task_job_id).unwrap_or(false); + let failed = result.is_err() && !cancelled; match (result, cancelled) { (_, true) => { - let _ = db.finish_item( - &job_id, + let _ = task_db.finish_item( + &task_job_id, &item.id, "cancelled", "cancelled", @@ -186,13 +293,18 @@ fn spawn_job( ); } (Ok(()), false) => { - let _ = - db.finish_item(&job_id, &item.id, "succeeded", "completed", duration, None); + let _ = task_db.finish_item( + &task_job_id, + &item.id, + "succeeded", + "completed", + duration, + None, + ); } (Err(error), false) => { - failed = true; - let _ = db.finish_item( - &job_id, + let _ = task_db.finish_item( + &task_job_id, &item.id, "failed", "failed", @@ -201,19 +313,37 @@ fn spawn_job( ); } } - emit_job(&app, &db, &job_id); - } + task_runtime.unregister_worker(&task_job_id, &item.id); + emit_job(&task_app, &task_db, &task_job_id); + failed + }); + } - if db.job_cancel_requested(&job_id).unwrap_or(false) { - let _ = db.cancel_queued_items(&job_id); - let _ = db.finish_job(&job_id, "cancelled", None); - } else if failed { - let _ = db.finish_job(&job_id, "failed", Some("部分条目执行失败")); - } else { - let _ = db.finish_job(&job_id, "succeeded", None); - } - emit_job(&app, &db, &job_id); - }); + let mut failed = false; + while let Some(result) = tasks.join_next().await { + failed |= result.unwrap_or(true); + } + finish_job(&app, &db, &job_id, failed); +} + +fn fail_all_analysis_items(app: &AppHandle, db: &Database, job: &Job, error: &str) { + for item in &job.items { + let _ = db.finish_item(&job.id, &item.id, "failed", "failed", 0, Some(error)); + } + let _ = db.finish_job(&job.id, "failed", Some(error)); + emit_job(app, db, &job.id); +} + +fn finish_job(app: &AppHandle, db: &Database, job_id: &str, failed: bool) { + if db.job_cancel_requested(job_id).unwrap_or(false) { + let _ = db.cancel_queued_items(job_id); + let _ = db.finish_job(job_id, "cancelled", None); + } else if failed { + let _ = db.finish_job(job_id, "failed", Some("部分条目执行失败")); + } else { + let _ = db.finish_job(job_id, "succeeded", None); + } + emit_job(app, db, job_id); } pub(crate) fn start_sync_internal( @@ -221,7 +351,8 @@ pub(crate) fn start_sync_internal( db: Arc, config: AppConfig, sidecar: Option>, - job_lock: Arc>, + job_lock: Arc>, + analysis_runtime: Arc, scope: Option<&str>, ) -> Result { let sources = config @@ -243,7 +374,15 @@ pub(crate) fn start_sync_internal( let job = db .create_job("sync", "queued", None, &items) .map_err(|error| error.to_string())?; - spawn_job(app, db, config, sidecar, job_lock, job.id.clone()); + spawn_job( + app, + db, + config, + sidecar, + job_lock, + analysis_runtime, + job.id.clone(), + ); Ok(job) } @@ -252,7 +391,8 @@ fn start_analysis_internal( db: Arc, config: AppConfig, sidecar: Option>, - job_lock: Arc>, + job_lock: Arc>, + analysis_runtime: Arc, session_ids: &[String], provider_profile_id: Option<&str>, ) -> Result { @@ -287,7 +427,15 @@ fn start_analysis_internal( &items, ) .map_err(|error| error.to_string())?; - spawn_job(app, db, config, sidecar, job_lock, job.id.clone()); + spawn_job( + app, + db, + config, + sidecar, + job_lock, + analysis_runtime, + job.id.clone(), + ); Ok(job) } @@ -303,6 +451,7 @@ pub async fn start_sync( state.config_snapshot(), state.sidecar.clone(), state.job_lock.clone(), + state.analysis_runtime.clone(), scope.as_deref(), ) } @@ -320,6 +469,7 @@ pub async fn start_analysis( state.config_snapshot(), state.sidecar.clone(), state.job_lock.clone(), + state.analysis_runtime.clone(), &session_ids, provider_profile_id.as_deref(), ) @@ -352,13 +502,12 @@ pub async fn cancel_job( .db .get_job(&job_id) .map_err(|error| error.to_string())?; - if job.kind == "analysis" && job.items.iter().any(|item| item.status == "running") { - if let Some(sidecar) = state.sidecar.clone() { - tokio::task::spawn_blocking(move || sidecar.stop()) - .await - .map_err(|error| error.to_string())? - .map_err(|error| error.to_string())?; - } + if job.kind == "analysis" { + let runtime = state.analysis_runtime.clone(); + let stopping_job_id = job_id.clone(); + tokio::task::spawn_blocking(move || runtime.stop_job(&stopping_job_id)) + .await + .map_err(|error| error.to_string())??; } emit_job(&app, &state.db, &job_id); state.db.get_job(&job_id).map_err(|error| error.to_string()) @@ -391,6 +540,7 @@ pub async fn retry_job_item( state.config_snapshot(), state.sidecar.clone(), state.job_lock.clone(), + state.analysis_runtime.clone(), item.source_id.as_deref(), ) } else { @@ -404,6 +554,7 @@ pub async fn retry_job_item( state.config_snapshot(), state.sidecar.clone(), state.job_lock.clone(), + state.analysis_runtime.clone(), &[session_id], profile_id, ) diff --git a/apps/desktop/src-tauri/src/config.rs b/apps/desktop/src-tauri/src/config.rs index 7869cbd..2b0fb2d 100644 --- a/apps/desktop/src-tauri/src/config.rs +++ b/apps/desktop/src-tauri/src/config.rs @@ -35,6 +35,8 @@ pub struct SourceConfig { pub struct DistillerConfig { #[serde(default)] pub active_profile_id: String, + #[serde(default = "default_max_concurrent_analyses")] + pub max_concurrent_analyses: usize, #[serde(default)] pub profiles: Vec, /// 仅用于把 v0.1 的单 API 配置迁移成 profile;落盘时移除。 @@ -56,6 +58,7 @@ impl Default for DistillerConfig { fn default() -> Self { Self { active_profile_id: "default".to_string(), + max_concurrent_analyses: default_max_concurrent_analyses(), profiles: vec![ProviderProfile { id: "default".to_string(), name: "OpenAI".to_string(), @@ -112,6 +115,9 @@ fn default_true() -> bool { fn default_timeout() -> u64 { 120 } +fn default_max_concurrent_analyses() -> usize { + 2 +} fn default_provider_kind() -> String { "api".to_string() } @@ -254,6 +260,7 @@ impl AppConfig { self.distiller = if let Some(api) = self.distiller.legacy_api.take() { DistillerConfig { active_profile_id: "migrated-api".to_string(), + max_concurrent_analyses: default_max_concurrent_analyses(), profiles: vec![ProviderProfile { id: "migrated-api".to_string(), name: "已迁移 API".to_string(), @@ -286,6 +293,11 @@ impl AppConfig { } pub fn validate(&self) -> ConfigResult<()> { + if !(1..=8).contains(&self.distiller.max_concurrent_analyses) { + return Err(ConfigError::Invalid( + "并行分析数必须在 1–8 之间".to_string(), + )); + } if self.distiller.profiles.is_empty() { return Err(ConfigError::Invalid( "至少需要一套分析 Provider 配置".to_string(), @@ -466,6 +478,7 @@ mod tests { ); assert_eq!(config.distiller.profiles.len(), 1); assert_eq!(config.distiller.active_profile_id, "default"); + assert_eq!(config.distiller.max_concurrent_analyses, 2); assert!(config.sync.scan_on_startup); } @@ -532,6 +545,15 @@ mod tests { assert!(config.validate().is_err()); } + #[test] + fn rejects_analysis_concurrency_outside_safe_range() { + let mut config = AppConfig::default(); + config.distiller.max_concurrent_analyses = 0; + assert!(config.validate().is_err()); + config.distiller.max_concurrent_analyses = 9; + assert!(config.validate().is_err()); + } + #[test] fn migrates_v01_single_api() { let text = r#" diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index a7851fd..c9ee7b2 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -6,8 +6,11 @@ mod path_local; mod sidecar; mod storage; -use std::sync::{Arc, RwLock}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; use tauri::Manager; +use tokio::sync::Notify; use config::AppConfig; use sidecar::SidecarManager; @@ -24,7 +27,93 @@ pub struct AppState { pub db: Arc, pub config: Arc>, pub sidecar: Option>, - pub job_lock: Arc>, + pub job_lock: Arc>, + pub analysis_runtime: Arc, +} + +pub struct AnalysisRuntime { + limit: AtomicUsize, + active: Mutex, + notify: Notify, + workers: Mutex>>>, +} + +pub struct AnalysisPermit { + runtime: Arc, +} + +impl AnalysisRuntime { + pub fn new(limit: usize) -> Self { + Self { + limit: AtomicUsize::new(limit), + active: Mutex::new(0), + notify: Notify::new(), + workers: Mutex::new(HashMap::new()), + } + } + + pub fn set_limit(&self, limit: usize) { + self.limit.store(limit, Ordering::Release); + self.notify.notify_waiters(); + } + + pub async fn acquire(self: &Arc) -> AnalysisPermit { + loop { + let notified = self.notify.notified(); + { + let mut active = self.active.lock().expect("analysis active lock poisoned"); + if *active < self.limit.load(Ordering::Acquire) { + *active += 1; + return AnalysisPermit { + runtime: self.clone(), + }; + } + } + notified.await; + } + } + + pub fn register_worker(&self, job_id: &str, item_id: &str, worker: Arc) { + self.workers + .lock() + .expect("analysis workers lock poisoned") + .entry(job_id.to_string()) + .or_default() + .insert(item_id.to_string(), worker); + } + + pub fn unregister_worker(&self, job_id: &str, item_id: &str) { + let mut workers = self.workers.lock().expect("analysis workers lock poisoned"); + if let Some(items) = workers.get_mut(job_id) { + items.remove(item_id); + if items.is_empty() { + workers.remove(job_id); + } + } + } + + pub fn stop_job(&self, job_id: &str) -> Result<(), String> { + let workers = self + .workers + .lock() + .map_err(|_| "分析任务锁异常,请重启应用".to_string())? + .get(job_id) + .map(|items| items.values().cloned().collect::>()) + .unwrap_or_default(); + for worker in workers { + worker.cancel().map_err(|error| error.to_string())?; + } + Ok(()) + } +} + +impl Drop for AnalysisPermit { + fn drop(&mut self) { + if let Ok(mut active) = self.runtime.active.lock() { + *active = active.saturating_sub(1); + } + self.runtime.notify.notify_one(); + } } impl AppState { @@ -73,11 +162,15 @@ pub fn run() { log::warn!("Sidecar 未就绪:提炼功能将不可用,直至构建或安装 chattake-sidecar"); } + let analysis_runtime = Arc::new(AnalysisRuntime::new( + config.distiller.max_concurrent_analyses, + )); let state = AppState { db: Arc::new(db), config: Arc::new(RwLock::new(config)), sidecar, - job_lock: Arc::new(tokio::sync::Mutex::new(())), + job_lock: Arc::new(tokio::sync::RwLock::new(())), + analysis_runtime, }; tauri::Builder::default() @@ -94,6 +187,7 @@ pub fn run() { state.config_snapshot(), state.sidecar.clone(), state.job_lock.clone(), + state.analysis_runtime.clone(), None, ) { log::error!("启动扫描任务创建失败: {}", error); @@ -138,3 +232,26 @@ pub fn run() { .run(context) .expect("Tauri 应用启动失败"); } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn analysis_runtime_applies_limit_updates() { + let runtime = Arc::new(AnalysisRuntime::new(1)); + let first = runtime.acquire().await; + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), runtime.acquire()) + .await + .is_err() + ); + + runtime.set_limit(2); + let second = tokio::time::timeout(std::time::Duration::from_millis(100), runtime.acquire()) + .await + .expect("提高并发上限后应立即获得许可"); + + drop((first, second)); + } +} diff --git a/apps/desktop/src-tauri/src/sidecar/mod.rs b/apps/desktop/src-tauri/src/sidecar/mod.rs index a605d3f..b445460 100644 --- a/apps/desktop/src-tauri/src/sidecar/mod.rs +++ b/apps/desktop/src-tauri/src/sidecar/mod.rs @@ -16,7 +16,8 @@ pub mod rpc; use std::path::PathBuf; use std::process::{Child, Command, Stdio}; -use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; use rpc::{RpcClient, RpcError}; use serde::de::DeserializeOwned; @@ -29,20 +30,37 @@ pub struct SidecarManager { /// 运行中的子进程 process: Mutex>, /// RPC 客户端(进程启动后创建) - client: Mutex>, + client: Mutex>>, /// sidecar 二进制路径 binary_path: PathBuf, + /// 任务取消后禁止 RPC 自动重启该 worker。 + cancelled: AtomicBool, + /// 独立分析 worker 退出后必须让当前条目失败,不能无 init 自动重启。 + restartable: bool, + started: AtomicBool, } impl SidecarManager { pub fn new(binary_path: PathBuf) -> Self { + Self::with_restart(binary_path, true) + } + + fn with_restart(binary_path: PathBuf, restartable: bool) -> Self { Self { process: Mutex::new(None), client: Mutex::new(None), binary_path, + cancelled: AtomicBool::new(false), + restartable, + started: AtomicBool::new(false), } } + /// 为一个分析条目创建独立进程管理器,隔离超时、取消与 Provider 状态。 + pub fn worker(&self) -> Self { + Self::with_restart(self.binary_path.clone(), false) + } + /// 查找 sidecar 二进制路径。 /// /// `package_info` 来自 [`tauri::generate_context!`],用于与 Tauri CLI 相同的 `resource_dir` 解析(跨平台)。 @@ -124,10 +142,16 @@ impl SidecarManager { .lock() .map_err(|_| RpcError::Internal("process lock poisoned".into()))?; + if self.cancelled.load(Ordering::Acquire) { + return Err(RpcError::Internal("Sidecar 任务已取消".into())); + } if proc_guard.is_some() { log::debug!("Sidecar 已在运行中"); return Ok(()); } + if self.started.load(Ordering::Acquire) && !self.restartable { + return Err(RpcError::Internal("Sidecar worker 已退出".into())); + } log::info!("启动 sidecar: {}", self.binary_path.display()); let mut child = Command::new(&self.binary_path) @@ -154,12 +178,19 @@ impl SidecarManager { .client .lock() .map_err(|_| RpcError::Internal("client lock poisoned".into()))?; - *client_guard = Some(rpc_client); + *client_guard = Some(Arc::new(rpc_client)); + self.started.store(true, Ordering::Release); log::info!("Sidecar 启动成功"); Ok(()) } + /// 取消任务并永久停止当前 worker,避免并发竞态触发自动重启。 + pub fn cancel(&self) -> Result<(), RpcError> { + self.cancelled.store(true, Ordering::Release); + self.stop() + } + /// 停止 sidecar 进程 pub fn stop(&self) -> Result<(), RpcError> { let mut proc_guard = self @@ -246,13 +277,12 @@ impl SidecarManager { self.start()?; } - let client_guard = self + let client = self .client .lock() - .map_err(|_| RpcError::Internal("client lock poisoned".into()))?; - - let client = client_guard + .map_err(|_| RpcError::Internal("client lock poisoned".into()))? .as_ref() + .cloned() .ok_or_else(|| RpcError::Internal("RPC 客户端未就绪".into()))?; client.call(method, params) @@ -269,16 +299,14 @@ impl SidecarManager { self.start()?; } - let result = { - let client_guard = self - .client - .lock() - .map_err(|_| RpcError::Internal("client lock poisoned".into()))?; - let client = client_guard - .as_ref() - .ok_or_else(|| RpcError::Internal("RPC 客户端未就绪".into()))?; - client.call_with_timeout(method, params, timeout) - }; + let client = self + .client + .lock() + .map_err(|_| RpcError::Internal("client lock poisoned".into()))? + .as_ref() + .cloned() + .ok_or_else(|| RpcError::Internal("RPC 客户端未就绪".into()))?; + let result = client.call_with_timeout(method, params, timeout); if matches!(result, Err(RpcError::Timeout(_))) { let _ = self.stop(); } @@ -291,3 +319,27 @@ impl Drop for SidecarManager { let _ = self.stop(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cancelled_worker_cannot_restart() { + let worker = SidecarManager::new(PathBuf::from("missing-sidecar")); + worker.cancel().unwrap(); + assert!( + matches!(worker.start(), Err(RpcError::Internal(message)) if message.contains("已取消")) + ); + } + + #[test] + fn analysis_worker_does_not_restart_without_init() { + let root = SidecarManager::new(PathBuf::from("missing-sidecar")); + let worker = root.worker(); + worker.started.store(true, Ordering::Release); + assert!( + matches!(worker.start(), Err(RpcError::Internal(message)) if message.contains("已退出")) + ); + } +} diff --git a/apps/desktop/src-tauri/src/sidecar/rpc.rs b/apps/desktop/src-tauri/src/sidecar/rpc.rs index 4ce3bb4..75bdb81 100644 --- a/apps/desktop/src-tauri/src/sidecar/rpc.rs +++ b/apps/desktop/src-tauri/src/sidecar/rpc.rs @@ -5,23 +5,24 @@ //! //! 注意:call() 是同步阻塞的,Tauri command 层需用 spawn_blocking 包装。 +use std::collections::HashMap; use std::io::{BufRead, BufReader, Write}; use std::process::{ChildStdin, ChildStdout}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{mpsc, Mutex}; -use std::time::{Duration, Instant}; +use std::sync::{mpsc, Arc, Mutex}; +use std::time::Duration; use serde::de::DeserializeOwned; use serde_json::Value; /// 默认 RPC 调用超时(秒),LLM 提炼可能需要较长时间 const DEFAULT_TIMEOUT_SECS: u64 = 120; +type PendingResponses = Arc>>>>; /// JSON-RPC 2.0 客户端,持有 sidecar 进程的 stdin/stdout pub struct RpcClient { - calls: Mutex<()>, stdin: Mutex, - responses: Mutex>>, + pending: PendingResponses, request_id: AtomicU64, /// RPC 调用默认超时时间 timeout: Duration, @@ -29,32 +30,32 @@ pub struct RpcClient { impl RpcClient { pub fn new(stdin: ChildStdin, stdout: ChildStdout) -> Self { - let (tx, rx) = mpsc::channel(); + let pending = PendingResponses::default(); + let reader_pending = pending.clone(); std::thread::spawn(move || { let mut reader = BufReader::new(stdout); loop { let mut line = String::new(); match reader.read_line(&mut line) { Ok(0) => { - let _ = tx.send(Err("Sidecar stdout 已关闭".into())); + fail_pending(&reader_pending, "Sidecar stdout 已关闭"); break; } Ok(_) => { - if tx.send(Ok(line)).is_err() { - break; + if let Err(error) = dispatch_response(&reader_pending, line) { + log::warn!("Sidecar 响应分发失败: {error}"); } } Err(error) => { - let _ = tx.send(Err(error.to_string())); + fail_pending(&reader_pending, &error.to_string()); break; } } } }); Self { - calls: Mutex::new(()), stdin: Mutex::new(stdin), - responses: Mutex::new(rx), + pending, request_id: AtomicU64::new(1), timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS), } @@ -80,11 +81,6 @@ impl RpcClient { params: Value, timeout: Duration, ) -> Result { - // Sidecar 允许并发处理请求,但桌面端当前不需要并发;整体串行可避免响应串线。 - let _call_guard = self - .calls - .lock() - .map_err(|_| RpcError::Internal("call lock poisoned".into()))?; let id = self.request_id.fetch_add(1, Ordering::SeqCst); let request = serde_json::json!({ @@ -104,42 +100,44 @@ impl RpcClient { timeout.as_secs() ); + let (response_tx, response_rx) = mpsc::channel(); + self.pending + .lock() + .map_err(|_| RpcError::Internal("pending response lock poisoned".into()))? + .insert(id, response_tx); + // 写 stdin - { + let write_result = (|| -> Result<(), RpcError> { let mut stdin = self .stdin .lock() .map_err(|_| RpcError::Internal("stdin lock poisoned".into()))?; writeln!(stdin, "{}", request_str).map_err(|e| RpcError::Io(e.to_string()))?; - stdin.flush().map_err(|e| RpcError::Io(e.to_string()))?; + stdin.flush().map_err(|e| RpcError::Io(e.to_string())) + })(); + if let Err(error) = write_result { + self.remove_pending(id); + return Err(error); } - // 超时请求的迟到响应可能仍在队列中;按 id 丢弃,直到拿到本次响应。 - let deadline = Instant::now() + timeout; - let response = loop { - let remaining = deadline.saturating_duration_since(Instant::now()); - if remaining.is_zero() { + let response_str = match response_rx.recv_timeout(timeout) { + Ok(Ok(line)) => line, + Ok(Err(error)) => return Err(RpcError::Io(error)), + Err(mpsc::RecvTimeoutError::Timeout) => { + self.remove_pending(id); return Err(RpcError::Timeout(timeout.as_secs())); } - let response_str = self.read_response_with_timeout(remaining)?; - if response_str.trim().is_empty() { - return Err(RpcError::Io("Sidecar 返回空响应(进程可能已退出)".into())); - } - let response: Value = serde_json::from_str(response_str.trim()).map_err(|e| { - RpcError::Deserialize(format!( - "响应解析失败: {} - {}", - e, - &response_str[..response_str.len().min(200)] - )) - })?; - if response.get("id").and_then(Value::as_u64) == Some(id) { - break response; + Err(mpsc::RecvTimeoutError::Disconnected) => { + return Err(RpcError::Io("读取线程异常退出".into())); } - log::warn!( - "丢弃非当前 RPC 响应: expected_id={id}, actual_id={:?}", - response.get("id") - ); }; + let response: Value = serde_json::from_str(response_str.trim()).map_err(|e| { + RpcError::Deserialize(format!( + "响应解析失败: {} - {}", + e, + &response_str[..response_str.len().min(200)] + )) + })?; // 检查 error 字段 if let Some(err) = response.get("error") { @@ -160,25 +158,64 @@ impl RpcClient { .map_err(|e| RpcError::Deserialize(format!("result 反序列化失败: {}", e))) } - /// 带超时的 stdout 读取。 - /// - /// stdout 由常驻读取线程持有;当前线程只在 channel 上等待,因此超时会立即返回。 - fn read_response_with_timeout(&self, timeout: Duration) -> Result { - let responses = self - .responses - .lock() - .map_err(|_| RpcError::Internal("response lock poisoned".into()))?; - match responses.recv_timeout(timeout) { - Ok(Ok(line)) => Ok(line), - Ok(Err(error)) => Err(RpcError::Io(error)), - Err(mpsc::RecvTimeoutError::Timeout) => Err(RpcError::Timeout(timeout.as_secs())), - Err(mpsc::RecvTimeoutError::Disconnected) => { - Err(RpcError::Io("读取线程异常退出".into())) - } + fn remove_pending(&self, id: u64) { + if let Ok(mut pending) = self.pending.lock() { + pending.remove(&id); } } } +fn dispatch_response(pending: &PendingResponses, line: String) -> Result<(), String> { + let response: Value = serde_json::from_str(line.trim()).map_err(|error| error.to_string())?; + let id = response + .get("id") + .and_then(Value::as_u64) + .ok_or_else(|| "响应缺少数字 id".to_string())?; + let sender = pending + .lock() + .map_err(|_| "pending response lock poisoned".to_string())? + .remove(&id) + .ok_or_else(|| format!("响应 id={id} 已超时或不存在"))?; + sender + .send(Ok(line)) + .map_err(|_| format!("响应 id={id} 的接收端已关闭")) +} + +fn fail_pending(pending: &PendingResponses, error: &str) { + let senders = pending + .lock() + .map(|mut pending| { + pending + .drain() + .map(|(_, sender)| sender) + .collect::>() + }) + .unwrap_or_default(); + for sender in senders { + let _ = sender.send(Err(error.to_string())); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn dispatches_out_of_order_responses_to_matching_calls() { + let pending = PendingResponses::default(); + let (first_tx, first_rx) = mpsc::channel(); + let (second_tx, second_rx) = mpsc::channel(); + pending.lock().unwrap().insert(1, first_tx); + pending.lock().unwrap().insert(2, second_tx); + + dispatch_response(&pending, r#"{"jsonrpc":"2.0","id":2,"result":"b"}"#.into()).unwrap(); + dispatch_response(&pending, r#"{"jsonrpc":"2.0","id":1,"result":"a"}"#.into()).unwrap(); + + assert!(second_rx.recv().unwrap().unwrap().contains("\"b\"")); + assert!(first_rx.recv().unwrap().unwrap().contains("\"a\"")); + } +} + /// RPC 调用错误 #[derive(Debug, thiserror::Error)] pub enum RpcError { diff --git a/apps/desktop/src-tauri/src/storage/cards.rs b/apps/desktop/src-tauri/src/storage/cards.rs index f383122..1584dda 100644 --- a/apps/desktop/src-tauri/src/storage/cards.rs +++ b/apps/desktop/src-tauri/src/storage/cards.rs @@ -685,6 +685,10 @@ mod tests { tech_stack: &technologies, }) .unwrap(); + assert_eq!( + db.get_session(&session_id).unwrap().card_id, + Some(high_id.clone()) + ); let draft_id = db .insert_card(&NewCard { session_id: &session_id, diff --git a/apps/desktop/src-tauri/src/storage/jobs.rs b/apps/desktop/src-tauri/src/storage/jobs.rs index feaa15c..b863295 100644 --- a/apps/desktop/src-tauri/src/storage/jobs.rs +++ b/apps/desktop/src-tauri/src/storage/jobs.rs @@ -1,5 +1,6 @@ use chrono::Utc; use rusqlite::{params, OptionalExtension, Row}; +use std::collections::HashSet; use uuid::Uuid; use super::db::{Database, DbError, DbResult}; @@ -61,6 +62,29 @@ impl Database { let now = Utc::now().to_rfc3339(); let mut conn = self.conn(); let tx = conn.transaction()?; + if kind == "analysis" { + let mut seen = HashSet::new(); + for session_id in items.iter().filter_map(|item| item.session_id) { + if !seen.insert(session_id) { + return Err(DbError::Invalid(format!( + "同一会话不能重复加入分析任务:{session_id}" + ))); + } + let active = tx + .query_row( + "SELECT 1 FROM job_items WHERE session_id=?1 AND status IN ('queued','running') LIMIT 1", + params![session_id], + |_| Ok(()), + ) + .optional()? + .is_some(); + if active { + return Err(DbError::Invalid(format!( + "会话已在分析队列中:{session_id}" + ))); + } + } + } let (profile_id, provider, base_url, model) = profile .map(|value| (Some(value.0), Some(value.1), Some(value.2), Some(value.3))) .unwrap_or((None, None, None, None)); @@ -329,6 +353,100 @@ mod tests { assert!(!columns.iter().any(|column| column == "api_key")); } + #[test] + fn rejects_duplicate_active_analysis_sessions() { + let dir = tempfile::tempdir().unwrap(); + let db = Database::open(&dir.path().join("jobs.db")).unwrap(); + let session = db + .insert_session( + "codex", + "duplicate", + "local", + None, + None, + 0, + None, + "/tmp/x", + "2026-01-01T00:00:00Z", + "2026-01-01T00:00:00Z", + None, + ) + .unwrap(); + let items = [NewJobItem { + session_id: Some(&session), + source_id: None, + raw_path: None, + }]; + let duplicate_items = [ + NewJobItem { + session_id: Some(&session), + source_id: None, + raw_path: None, + }, + NewJobItem { + session_id: Some(&session), + source_id: None, + raw_path: None, + }, + ]; + let error = db + .create_job("analysis", "queued", None, &duplicate_items) + .unwrap_err(); + assert!(error.to_string().contains("重复加入")); + + db.create_job("analysis", "queued", None, &items).unwrap(); + + let error = db + .create_job("analysis", "queued", None, &items) + .unwrap_err(); + assert!(error.to_string().contains("已在分析队列中")); + } + + #[test] + fn cancelling_running_item_releases_session_for_retry() { + let dir = tempfile::tempdir().unwrap(); + let db = Database::open(&dir.path().join("jobs.db")).unwrap(); + let session = db + .insert_session( + "codex", + "cancelled", + "local", + None, + None, + 0, + None, + "/tmp/x", + "2026-01-01T00:00:00Z", + "2026-01-01T00:00:00Z", + None, + ) + .unwrap(); + let items = [NewJobItem { + session_id: Some(&session), + source_id: None, + raw_path: None, + }]; + let job = db.create_job("analysis", "queued", None, &items).unwrap(); + db.mark_job_running(&job.id, "judging").unwrap(); + db.mark_item_running(&job.items[0].id, "judging").unwrap(); + db.request_job_cancel(&job.id).unwrap(); + db.finish_item( + &job.id, + &job.items[0].id, + "cancelled", + "cancelled", + 0, + Some("用户取消"), + ) + .unwrap(); + db.finish_job(&job.id, "cancelled", None).unwrap(); + + let cancelled = db.get_job(&job.id).unwrap(); + assert_eq!(cancelled.done, cancelled.total); + assert_eq!(cancelled.items[0].status, "cancelled"); + assert!(db.create_job("analysis", "queued", None, &items).is_ok()); + } + #[test] fn progress_snapshot_caps_file_items() { let dir = tempfile::tempdir().unwrap(); diff --git a/apps/desktop/src-tauri/src/storage/models.rs b/apps/desktop/src-tauri/src/storage/models.rs index ee20a09..9d1369d 100644 --- a/apps/desktop/src-tauri/src/storage/models.rs +++ b/apps/desktop/src-tauri/src/storage/models.rs @@ -44,6 +44,9 @@ pub struct Session { pub analyzed_at: Option, /// 分析失败时的错误信息 pub error_message: Option, + /// 当前会话最新生成的知识卡片 + #[serde(default)] + pub card_id: Option, } /// 会话列表轻量版(列表所需路径 / 首条 user 预览等;不含 content_hash) diff --git a/apps/desktop/src-tauri/src/storage/sessions.rs b/apps/desktop/src-tauri/src/storage/sessions.rs index ce1189d..cd4b73f 100644 --- a/apps/desktop/src-tauri/src/storage/sessions.rs +++ b/apps/desktop/src-tauri/src/storage/sessions.rs @@ -246,7 +246,9 @@ impl Database { conn.query_row( "SELECT id, source_id, external_session_id, source_host, project_path, project_name, message_count, content_hash, raw_path, created_at, updated_at, - status, value, has_updates, analyzed_at, error_message, analysis_title + status, value, has_updates, analyzed_at, error_message, analysis_title, + (SELECT c.id FROM cards c WHERE c.session_id = sessions.id + ORDER BY c.created_at DESC, c.id DESC LIMIT 1) FROM sessions WHERE id = ?1", params![id], |row| { @@ -268,6 +270,7 @@ impl Database { analyzed_at: row.get(14)?, error_message: row.get(15)?, analysis_title: row.get(16)?, + card_id: row.get(17)?, }) }, ) diff --git a/apps/desktop/src/components/AnalysisQueuePanel.vue b/apps/desktop/src/components/AnalysisQueuePanel.vue index 3455f05..b9aba02 100644 --- a/apps/desktop/src/components/AnalysisQueuePanel.vue +++ b/apps/desktop/src/components/AnalysisQueuePanel.vue @@ -1,5 +1,5 @@