diff --git a/crates/df-ai/Cargo.toml b/crates/df-ai/Cargo.toml index d02b327..08502f0 100644 --- a/crates/df-ai/Cargo.toml +++ b/crates/df-ai/Cargo.toml @@ -7,7 +7,7 @@ edition = "2021" df-core = { path = "../df-core" } serde = { workspace = true } serde_json = { workspace = true } -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["sync", "time"] } async-trait = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index 8c501fa..2e2bcd9 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -11,19 +11,23 @@ use eventsource_stream::Eventsource; use futures::StreamExt; use reqwest::Client; use serde::{Deserialize, Serialize}; +use std::time::Duration; use tracing::{debug, error, warn}; use crate::provider::{ CompletionRequest, CompletionResponse, LlmProvider, MessageRole, StreamChunk, StreamResult, TokenUsage, ToolCall, ToolCallDelta, }; +use crate::retry::{ + retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable, +}; // ============================================================ // Anthropic API 请求/响应结构体 // ============================================================ /// Anthropic 请求体 -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] struct AnthropicRequest { model: String, messages: Vec, @@ -40,7 +44,7 @@ struct AnthropicRequest { } /// Anthropic 工具定义（input_schema 对应 OpenAI 的 parameters） -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] struct AnthropicToolDef { name: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -406,68 +410,93 @@ impl LlmProvider for AnthropicCompatProvider { debug!(model = %body.model, "Anthropic 同步调用"); - let resp = self - .auth_headers(self.client.post(self.messages_url())) - .json(&body) - .send() - .await?; - - if !resp.status().is_success() { - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - error!(%status, %text, "Anthropic API 调用失败"); - anyhow::bail!("Anthropic API 错误 {}: {}", status, text); - } - - let resp: AnthropicResponse = resp.json().await?; - - // content 块中拼接 text，收集 tool_use - let mut text = String::new(); - let mut tool_calls: Vec = Vec::new(); - for block in resp.content { - match block.block_type.as_str() { - "text" => { - if let Some(t) = block.text { - text.push_str(&t); + // 指数退避重试(B-260616-07): 包裹 send + 状态码判定。 + // 同时补 FR-R4 遗漏: Anthropic 同步路径此前无单请求 timeout(建连后挂起会无限 hang)， + // 此处加 60s timeout，与 OpenAI 路径对齐。 + let label = format!("Anthropic[{}]", body.model); + retry_with_backoff(&label, move |_| { + let client = self.client.clone(); + let url = self.messages_url(); + let api_key = self.api_key.clone(); + let body = body.clone(); + async move { + // send: 补 60s 单请求 timeout(此前缺失，FR-R4 遗漏) + let rb = client + .post(url) + .header("x-api-key", &api_key) + .header("anthropic-version", ANTHROPIC_VERSION) + .header("Content-Type", "application/json") + .timeout(Duration::from_secs(60)) + .json(&body); + let resp = match rb.send().await { + Ok(r) => r, + Err(e) => { + if is_reqwest_error_retryable(&e) { + return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); + } + return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", e)); + } + }; + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let text = resp.text().await.unwrap_or_default(); + let msg = format!("Anthropic API 错误 {}: {}", status, text); + if is_status_retryable(status) { + warn!(%status, "Anthropic 同步调用可重试状态码"); + return AttemptOutcome::Retryable(msg); + } + error!(%status, %text, "Anthropic API 调用失败(不可重试)"); + return AttemptOutcome::Fatal(msg); + } + let resp: AnthropicResponse = match resp.json().await { + Ok(r) => r, + Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + }; + // content 块中拼接 text，收集 tool_use + let mut text = String::new(); + let mut tool_calls: Vec = Vec::new(); + for block in resp.content { + match block.block_type.as_str() { + "text" => { + if let Some(t) = block.text { + text.push_str(&t); + } + } + "tool_use" => { + let id = match block.id { + Some(id) if !id.is_empty() => id, + _ => { + warn!( + name = ?block.name, + "Anthropic tool_use 块缺少 id，已跳过（空 id 会回传空 tool_use_id 触发 500）" + ); + continue; + } + }; + let name = block.name.unwrap_or_default(); + let args = block + .input + .map(|v| serde_json::to_string(&v).unwrap_or_default()) + .unwrap_or_default(); + tool_calls.push(ToolCall::new(id, name, args)); + } + other => warn!(block_type = other, "Anthropic 响应含未知 content 块类型，已忽略"), } } - "tool_use" => { - // tool_use 必须带 id：后续 tool_result 用它回引，缺 id 的 tool_call - // 会拖出空 tool_use_id 的 tool_result，触发 GLM 500 卡死会话。 - // id 缺失/为空时跳过该 tool_use 解析并告警。 - let id = match block.id { - Some(id) if !id.is_empty() => id, - _ => { - warn!( - name = ?block.name, - "Anthropic tool_use 块缺少 id，已跳过（空 id 会回传空 tool_use_id 触发 500）" - ); - continue; - } - }; - let name = block.name.unwrap_or_default(); - let args = block - .input - .map(|v| serde_json::to_string(&v).unwrap_or_default()) - .unwrap_or_default(); - tool_calls.push(ToolCall::new(id, name, args)); - } - other => warn!(block_type = other, "Anthropic 响应含未知 content 块类型，已忽略"), + let usage = TokenUsage { + prompt_tokens: resp.usage.input_tokens, + completion_tokens: resp.usage.output_tokens, + total_tokens: resp.usage.input_tokens + resp.usage.output_tokens, + }; + AttemptOutcome::Ok(CompletionResponse { + text, + model: resp.model, + usage, + tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) }, + }) } - } - - let usage = TokenUsage { - prompt_tokens: resp.usage.input_tokens, - completion_tokens: resp.usage.output_tokens, - total_tokens: resp.usage.input_tokens + resp.usage.output_tokens, - }; - - Ok(CompletionResponse { - text, - model: resp.model, - usage, - tool_calls: if tool_calls.is_empty() { None } else { Some(tool_calls) }, }) + .await } async fn stream(&self, request: CompletionRequest) -> anyhow::Result { diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index ce972b3..0e2e2ce 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -6,6 +6,7 @@ pub mod context; pub mod coordinator; pub mod openai_compat; pub mod provider; +mod retry; use provider::LlmProvider; diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 19db946..1cddf74 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -16,13 +16,16 @@ use crate::provider::{ CompletionRequest, CompletionResponse, LlmProvider, StreamChunk, StreamResult, TokenUsage, ToolCall, ToolCallDelta, }; +use crate::retry::{ + retry_with_backoff, AttemptOutcome, is_reqwest_error_retryable, is_status_retryable, +}; // ============================================================ // OpenAI API 请求/响应结构体 // ============================================================ /// OpenAI 兼容请求体 -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] struct OpenAiRequest { model: String, messages: Vec, @@ -41,7 +44,7 @@ struct OpenAiRequest { } /// OpenAI 消息格式 -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] struct OpenAiMessage { role: String, content: String, @@ -398,60 +401,75 @@ impl LlmProvider for OpenAICompatProvider { debug!(model = %openai_req.model, "OpenAI 同步调用"); - // 同步路径整体超时(FR-R4)：client 仅配 connect_timeout，无总 timeout（防误砍流式长生成）。 - // complete 为一次性请求，挂死会让整轮对话静默卡住，此处用 RequestBuilder 的单请求 timeout 兜底， - // 超时返回 is_timeout 错误（不静默挂），且不影响 stream() 流式路径。 - let resp = self - .client - .post(self.chat_url()) - .header("Authorization", format!("Bearer {}", self.api_key)) - .header("Content-Type", "application/json") - .timeout(Duration::from_secs(60)) - .json(&openai_req) - .send() - .await - .map_err(|e| { - if e.is_timeout() { - error!(model = %openai_req.model, "OpenAI 同步调用超时(60s)"); - anyhow::anyhow!("LLM 同步调用超时(60s)，请检查网络或模型可用性: {}", e) - } else { - anyhow::anyhow!(e) + // 指数退避重试(B-260616-07): 包裹 send + 状态码判定。 + // 单请求 60s timeout 保持不变(FR-R4)，重试是额外层: 3 次 × 60s 最坏 180s， + // 由 retry_with_backoff 内部 30s 总预算主动止损。 + let label = format!("OpenAI[{}]", openai_req.model); + retry_with_backoff(&label, move |_| { + let client = self.client.clone(); + let url = self.chat_url(); + let api_key = self.api_key.clone(); + let openai_req = openai_req.clone(); + async move { + // send + let resp = client + .post(url) + .header("Authorization", format!("Bearer {}", api_key)) + .header("Content-Type", "application/json") + .timeout(Duration::from_secs(60)) + .json(&openai_req) + .send() + .await; + let resp = match resp { + Ok(r) => r, + Err(e) => { + if is_reqwest_error_retryable(&e) { + return AttemptOutcome::Retryable(format!("请求失败(可重试): {}", e)); + } + return AttemptOutcome::Fatal(format!("请求失败(不可重试): {}", e)); + } + }; + // 状态码判定 + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + let msg = format!("LLM API 错误 {}: {}", status, body); + if is_status_retryable(status) { + warn!(%status, "OpenAI 同步调用可重试状态码"); + return AttemptOutcome::Retryable(msg); + } + error!(%status, %body, "LLM API 调用失败(不可重试)"); + return AttemptOutcome::Fatal(msg); } - })?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - error!(%status, %body, "LLM API 调用失败"); - anyhow::bail!("LLM API 错误 {}: {}", status, body); - } - - let body: OpenAiResponse = resp.json().await?; - let choice = body - .choices - .into_iter() - .next() - .ok_or_else(|| anyhow::anyhow!("LLM 响应无 choices"))?; - - let text = choice.message.content.unwrap_or_default(); - let tool_calls = choice.message.tool_calls.map(Self::parse_tool_calls); - - let usage = body.usage.map(|u| TokenUsage { - prompt_tokens: u.prompt_tokens, - completion_tokens: u.completion_tokens, - total_tokens: u.total_tokens, - }).unwrap_or(TokenUsage { - prompt_tokens: 0, - completion_tokens: 0, - total_tokens: 0, - }); - - Ok(CompletionResponse { - text, - model: body.model, - usage, - tool_calls, + // body 解析: 解析错属 Fatal(响应已成功送达，重试也会因同样格式失败) + let body: OpenAiResponse = match resp.json().await { + Ok(b) => b, + Err(e) => return AttemptOutcome::Fatal(format!("响应解析失败: {}", e)), + }; + let choice = match body.choices.into_iter().next() { + Some(c) => c, + None => return AttemptOutcome::Fatal("LLM 响应无 choices".to_string()), + }; + let text = choice.message.content.unwrap_or_default(); + let tool_calls = choice.message.tool_calls.map(Self::parse_tool_calls); + let usage = body.usage.map(|u| TokenUsage { + prompt_tokens: u.prompt_tokens, + completion_tokens: u.completion_tokens, + total_tokens: u.total_tokens, + }).unwrap_or(TokenUsage { + prompt_tokens: 0, + completion_tokens: 0, + total_tokens: 0, + }); + AttemptOutcome::Ok(CompletionResponse { + text, + model: body.model, + usage, + tool_calls, + }) + } }) + .await } async fn stream(&self, request: CompletionRequest) -> anyhow::Result { diff --git a/crates/df-ai/src/retry.rs b/crates/df-ai/src/retry.rs new file mode 100644 index 0000000..5844c96 --- /dev/null +++ b/crates/df-ai/src/retry.rs @@ -0,0 +1,257 @@ +//! complete() 指数退避重试 — 仅非流式同步路径，不动 stream() +//! +//! 背景: complete() 是 LLM 一次性请求，网络波动/瞬态 5xx/429 会让整轮对话静默失败。 +//! 各调用方(title.rs/knowledge_inject.rs/project.rs/ai_node.rs)虽已有人工兜底降级， +//! 但重试能在不侵入上层的前提下吸收瞬态错误，对低频后台调用零风险。 +//! +//! 策略: +//! - 最多 3 次尝试(初次 + 2 次重试)，指数退避 1s → 2s → 4s + ±20% jitter +//! - 总挂钟预算 30s(含 sleep + 各次请求耗时)，超预算直接放弃重试返回最后错误 +//! - 仅对可重试错误重试: timeout / connect / 5xx / 429; 不重试 4xx(非429) / 参数错 / 鉴权错 +//! - 不引入 reqwest-retry / tower: complete 路径简单，手写 loop 零依赖、可调试、不与 SSE/双层 Semaphore 耦合 +//! +//! 流式 stream() 路径不进入此模块(流式长生成不能整体 timeout，重试会丢增量)。 +//! +//! 重试期间不释放 Semaphore permit(已在调用方持有)，对并发池有挤占 —— 但 complete 调用低频可接受。 + +use std::future::Future; +use std::time::{Duration, SystemTime}; + +use tracing::warn; + +/// 最多尝试次数(含初次)。B-260616-07: 3 次 = 初次 + 2 次重试。 +/// +/// 配置化 TODO: 未来接入 per-provider config(`AiProviderRecord.config` JSON)或全局开关 +/// (`useSetting('df-ai-max-retries')`)时改为读取配置。当前低频后台调用，常量足够。 +pub const MAX_COMPLETE_ATTEMPTS: u32 = 3; + +/// 退避基准(秒): 第 n 次重试前 sleep `BASE_BACKOFF_SECS * 2^(n-1)`。 +/// 即 1s → 2s → 4s。 +const BASE_BACKOFF_SECS: u64 = 1; + +/// 整体重试挂钟预算(含 sleep 与各次请求耗时)。超过即放弃，返回最后错误。 +/// 与单请求 60s timeout 配合: 最坏 3 次 × 60s = 180s，30s 预算会在此之前主动止损。 +const MAX_TOTAL_BUDGET: Duration = Duration::from_secs(30); + +/// jitter 上限(相对 base 的 ±比例)。避免重试风暴对齐。 +const JITTER_RATIO: f64 = 0.2; + +/// 一次尝试的分类结果 —— 在 anyhow 不透明化前决定是否值得重试。 +#[derive(Debug)] +pub(super) enum AttemptOutcome { + /// 成功，携带结果。 + Ok(T), + /// 可重试错误(timeout / connect / 5xx / 429)。携带人类可读错误描述。 + Retryable(String), + /// 不可重试错误(4xx 非 429 / 鉴权 / 参数错 / 解析错)。携带描述，直接放弃。 + Fatal(String), +} + +/// 判定 reqwest::Error 是否可重试(仅 timeout / connect 类)。 +/// +/// body 解析错(reqwest::Error::is_decode)与请求构造错(is_builder)属 Fatal，不重试。 +pub(super) fn is_reqwest_error_retryable(e: &reqwest::Error) -> bool { + e.is_timeout() || e.is_connect() || (e.is_request() && !e.is_builder()) +} + +/// 判定 HTTP 状态码是否可重试: 5xx 与 429 重试，其余(含 4xx)不重试。 +pub(super) fn is_status_retryable(status: u16) -> bool { + status == 429 || (500..600).contains(&status) +} + +/// 指数退避 + jitter: 返回第 `attempt`(1-based)次重试前应 sleep 的时长。 +/// attempt=1 → ~1s, attempt=2 → ~2s, attempt=3 → ~4s，各 ±20% jitter。 +/// +/// jitter 用 SystemTime 纳秒取模生成(无依赖)，避免多客户端同步重试风暴。 +/// 以毫秒粒度计算后向下取整(避免秒级截断把 0.9s 砍成 0)。 +fn backoff_delay(attempt: u32) -> Duration { + let base_ms = BASE_BACKOFF_SECS.saturating_mul(1u64 << (attempt - 1)) * 1000; + // 纳秒 → [0, 2000) 区间，再映射到 [-1.0, +1.0) 比例 + let nanos = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0); + let jitter_ratio = (nanos % 2000) as f64 / 1000.0 - 1.0; // [-1.0, 1.0) + let factor = 1.0 + jitter_ratio * JITTER_RATIO; // [0.8, 1.2] + let ms = (base_ms as f64 * factor).max(0.0) as u64; + Duration::from_millis(ms) +} + +/// 对一个返回 `AttemptOutcome` 的异步操作做指数退避重试。 +/// +/// `attempt_fn` 每次被调用执行一次尝试并返回分类结果。 +/// 达到 `MAX_COMPLETE_ATTEMPTS` 或超出 `MAX_TOTAL_BUDGET` 或遇 Fatal 即终止。 +/// 终止时若仍失败，返回最后一次的 Retryable/Fatal 描述作为 `anyhow::Error`。 +/// +/// 调用方约定: `attempt_fn` 内部应保留完整请求重发能力(每次重建 RequestBuilder)， +/// 因为 reqwest::RequestBuilder 一次 `.send()` 后不可复用。 +pub(super) async fn retry_with_backoff( + label: &str, + attempt_fn: F, +) -> anyhow::Result +where + F: Fn(u32) -> Fut, + Fut: Future>, +{ + let deadline = tokio::time::Instant::now() + MAX_TOTAL_BUDGET; + let mut last_err = String::from("未尝试"); + + for attempt in 1..=MAX_COMPLETE_ATTEMPTS { + match attempt_fn(attempt).await { + AttemptOutcome::Ok(v) => return Ok(v), + AttemptOutcome::Fatal(msg) => { + // 不可重试: 立即放弃，不浪费预算 + return Err(anyhow::anyhow!("{}", msg)); + } + AttemptOutcome::Retryable(msg) => { + last_err = msg; + let is_last = attempt == MAX_COMPLETE_ATTEMPTS; + if is_last { + warn!( + label = label, + attempt = attempt, + "重试已达上限({}次)，放弃", MAX_COMPLETE_ATTEMPTS + ); + break; + } + let now = tokio::time::Instant::now(); + if now >= deadline { + warn!( + label = label, + attempt = attempt, + "重试预算({:?})已耗尽，放弃", MAX_TOTAL_BUDGET + ); + break; + } + let delay = backoff_delay(attempt).min(deadline.saturating_duration_since(now)); + warn!( + label = label, + attempt = attempt, + delay_ms = delay.as_millis() as u64, + err = %last_err, + "可重试错误，退避后重试" + ); + tokio::time::sleep(delay).await; + } + } + } + + Err(anyhow::anyhow!("{} 重试耗尽: {}", label, last_err)) +} + +// ============================================================ +// 单测(纯逻辑，不发 HTTP) +// ============================================================ + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Arc; + + #[test] + fn backoff_is_monotonic_within_jitter() { + // base 序列: 1, 2, 4 秒; jitter ±20% 后仍在 [0.8,4.8] 范围 + for attempt in 1..=3 { + let d = backoff_delay(attempt); + let base_ms = (BASE_BACKOFF_SECS * (1u64 << (attempt - 1))) * 1000; + let lo = (base_ms as f64 * (1.0 - JITTER_RATIO)) as u64; + let hi = (base_ms as f64 * (1.0 + JITTER_RATIO)) as u64; + assert!( + d.as_millis() as u64 >= lo && d.as_millis() as u64 <= hi, + "attempt={} delay={:?} 应在 [{},{}]ms", + attempt, + d, + lo, + hi + ); + } + } + + #[test] + fn status_classification() { + assert!(is_status_retryable(429)); + assert!(is_status_retryable(500)); + assert!(is_status_retryable(503)); + assert!(!is_status_retryable(400)); + assert!(!is_status_retryable(401)); + assert!(!is_status_retryable(403)); + assert!(!is_status_retryable(404)); + } + + /// 初次成功 → 不重试 + #[tokio::test] + async fn first_attempt_ok() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_clone = calls.clone(); + let r = retry_with_backoff("test", move |_| { + let c = calls_clone.clone(); + async move { + c.fetch_add(1, Ordering::SeqCst); + AttemptOutcome::Ok(42u32) + } + }) + .await + .unwrap(); + assert_eq!(r, 42); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + /// 第二次成功 → 恰好 2 次调用 + #[tokio::test] + async fn retry_once_then_ok() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_clone = calls.clone(); + let r = retry_with_backoff("test", move |_| { + let c = calls_clone.clone(); + async move { + let n = c.fetch_add(1, Ordering::SeqCst) + 1; + if n == 1 { + AttemptOutcome::Retryable("timeout".into()) + } else { + AttemptOutcome::Ok(7u32) + } + } + }) + .await + .unwrap(); + assert_eq!(r, 7); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + /// 全部 Retryable → 用尽 3 次，返回错误 + #[tokio::test] + async fn all_retryable_exhausts() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_clone = calls.clone(); + let r = retry_with_backoff("test", move |_| { + let c = calls_clone.clone(); + async move { + c.fetch_add(1, Ordering::SeqCst); + AttemptOutcome::::Retryable("503".into()) + } + }) + .await; + assert!(r.is_err()); + assert!(r.unwrap_err().to_string().contains("重试耗尽")); + assert_eq!(calls.load(Ordering::SeqCst), MAX_COMPLETE_ATTEMPTS); + } + + /// Fatal 立即放弃，不重试 + #[tokio::test] + async fn fatal_aborts_immediately() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_clone = calls.clone(); + let r = retry_with_backoff("test", move |_| { + let c = calls_clone.clone(); + async move { + c.fetch_add(1, Ordering::SeqCst); + AttemptOutcome::::Fatal("401".into()) + } + }) + .await; + assert!(r.is_err()); + assert!(r.unwrap_err().to_string().contains("401")); + assert_eq!(calls.load(Ordering::SeqCst), 1, "Fatal 不应触发重试"); + } +} diff --git a/crates/df-execute/src/shell.rs b/crates/df-execute/src/shell.rs index 3353106..949d743 100644 --- a/crates/df-execute/src/shell.rs +++ b/crates/df-execute/src/shell.rs @@ -16,6 +16,24 @@ pub struct ShellResult { pub duration_ms: u64, } +/// Shell 类型选择 +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum ShellType { + /// Windows cmd.exe (默认 Windows) + Cmd, + /// PowerShell + PowerShell, + /// Unix sh (默认非 Windows) + Sh, +} + +impl Default for ShellType { + fn default() -> Self { + if cfg!(windows) { ShellType::Cmd } else { ShellType::Sh } + } +} + /// Shell 命令执行请求 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ShellRequest { @@ -27,6 +45,9 @@ pub struct ShellRequest { pub env: std::collections::HashMap, /// 超时时间（秒），None 表示不超时 pub timeout_secs: Option, + /// Shell 类型（默认：Windows→Cmd, 非Windows→Sh） + #[serde(default)] + pub shell_type: Option, } /// 执行 Shell 命令 @@ -35,16 +56,26 @@ pub struct ShellRequest { pub async fn execute(request: ShellRequest) -> anyhow::Result { let start = std::time::Instant::now(); - let mut cmd = if cfg!(windows) { - let mut c = tokio::process::Command::new("cmd"); - c.arg("/C").arg(&request.command); - c.stdout(Stdio::piped()).stderr(Stdio::piped()); - c - } else { - let mut c = tokio::process::Command::new("sh"); - c.arg("-c").arg(&request.command); - c.stdout(Stdio::piped()).stderr(Stdio::piped()); - c + let shell_type = request.shell_type.unwrap_or_default(); + let mut cmd = match shell_type { + ShellType::PowerShell => { + let mut c = tokio::process::Command::new("powershell"); + c.arg("-NoProfile").arg("-Command").arg(&request.command); + c.stdout(Stdio::piped()).stderr(Stdio::piped()); + c + } + ShellType::Cmd => { + let mut c = tokio::process::Command::new("cmd"); + c.arg("/C").arg(&request.command); + c.stdout(Stdio::piped()).stderr(Stdio::piped()); + c + } + ShellType::Sh => { + let mut c = tokio::process::Command::new("sh"); + c.arg("-c").arg(&request.command); + c.stdout(Stdio::piped()).stderr(Stdio::piped()); + c + } }; if let Some(dir) = &request.working_dir { diff --git a/src-tauri/src/commands/ai/audit.rs b/src-tauri/src/commands/ai/audit.rs index 06bbc81..24c04eb 100644 --- a/src-tauri/src/commands/ai/audit.rs +++ b/src-tauri/src/commands/ai/audit.rs @@ -69,59 +69,84 @@ async fn build_approval_reason( db: &Arc, ) -> String { let s = |key: &str| args.get(key).and_then(|v| v.as_str()).unwrap_or(""); - let detail = match tool_name { - "delete_project" => { - let id = s("id"); - if !id.is_empty() { format!("删除项目{}", resolve_project_label(db, id).await) } else { String::new() } + // 优先级：tool_display_hint(轻量标签) → display_hint_for_tool(模板填充) → 硬编码 fallback + let detail = if let Some(hint) = super::tool_registry::tool_display_hint(tool_name) { + // 轻量命中：直接用中文动作前缀 + 风险后缀（不含参数细节） + hint.to_string() + } else if let Some((template, keys)) = super::tool_registry::display_hint_for_tool(tool_name) { + // 按 keys 列表取参数值；含 "id"/"project_id" 的值走 resolve_project_label 解析 + let mut values = Vec::with_capacity(keys.len()); + for &key in keys { + let val = s(key); + if val.is_empty() { values.clear(); break; } + match key { + "id" | "project_id" => values.push(resolve_project_label(db, &val).await), + _ => values.push(val.to_string()), + } } - "restore_project" => { - let id = s("id"); - if !id.is_empty() { format!("从回收站恢复项目{}", resolve_project_label(db, id).await) } else { String::new() } + // 任一关键参数为空则整条 fallback 为空串（与原行为一致） + if values.is_empty() { + String::new() + } else { + // 模板 {} 占位符按位置填充 + let mut result = template.to_string(); + for v in &values { + result = result.replacen("{}", v, 1); + } + result } - "purge_project" => { - let id = s("id"); - if !id.is_empty() { format!("永久删除项目及关联数据，不可恢复{}", resolve_project_label(db, id).await) } else { String::new() } + } else { + // fallback：原有完整硬编码逻辑（保持行为不变） + match tool_name { + "delete_project" => { + let id = s("id"); + if !id.is_empty() { format!("删除项目{}", resolve_project_label(db, id).await) } else { String::new() } + } + "restore_project" => { + let id = s("id"); + if !id.is_empty() { format!("从回收站恢复项目{}", resolve_project_label(db, id).await) } else { String::new() } + } + "purge_project" => { + let id = s("id"); + if !id.is_empty() { format!("永久删除项目及关联数据，不可恢复{}", resolve_project_label(db, id).await) } else { String::new() } + } + "update_project" => { + let id = s("id"); + let field = s("field"); + if !id.is_empty() { + format!("修改项目{}字段「{}」", resolve_project_label(db, id).await, field) + } else if !field.is_empty() { + format!("修改项目字段「{}」", field) + } else { String::new() } + } + "bind_directory" => { + let id = s("id"); + let path = s("path"); + if !path.is_empty() && !id.is_empty() { + format!("绑定目录：{}（项目{}）", path, resolve_project_label(db, id).await) + } else if !path.is_empty() { + format!("绑定目录：{}", path) + } else { String::new() } + } + "create_task" => { + let title = s("title"); + let pid = s("project_id"); + if !title.is_empty() && !pid.is_empty() { + format!("创建任务：{}（项目{}）", title, resolve_project_label(db, pid).await) + } else if !title.is_empty() { + format!("创建任务：{}", title) + } else { String::new() } + } + "create_project" => { + let name = s("name"); + if !name.is_empty() { format!("创建项目：「{}」", name) } else { String::new() } + } + "create_idea" => { + let title = s("title"); + if !title.is_empty() { format!("捕获灵感：{}", title) } else { String::new() } + } + _ => String::new(), } - "update_project" => { - let id = s("id"); - let field = s("field"); - if !id.is_empty() { - format!("修改项目{}字段「{}」", resolve_project_label(db, id).await, field) - } else if !field.is_empty() { - format!("修改项目字段「{}」", field) - } else { String::new() } - } - "bind_directory" => { - let id = s("id"); - let path = s("path"); - if !path.is_empty() && !id.is_empty() { - format!("绑定目录：{}（项目{}）", path, resolve_project_label(db, id).await) - } else if !path.is_empty() { - format!("绑定目录：{}", path) - } else { String::new() } - } - "create_task" => { - let title = s("title"); - let pid = s("project_id"); - if !title.is_empty() && !pid.is_empty() { - format!("创建任务：{}（项目{}）", title, resolve_project_label(db, pid).await) - } else if !title.is_empty() { - format!("创建任务：{}", title) - } else { String::new() } - } - "create_project" => { - let name = s("name"); - if !name.is_empty() { format!("创建项目：「{}」", name) } else { String::new() } - } - "create_idea" => { - let title = s("title"); - if !title.is_empty() { format!("捕获灵感：{}", title) } else { String::new() } - } - "run_workflow" => { - let name = s("name"); - if !name.is_empty() { format!("运行工作流：{}", name) } else { String::new() } - } - _ => String::new(), }; if detail.is_empty() { // fallback：无可读字段，保留风险等级提示模板 diff --git a/src-tauri/src/commands/ai/commands.rs b/src-tauri/src/commands/ai/commands.rs index bbef506..9b54129 100644 --- a/src-tauri/src/commands/ai/commands.rs +++ b/src-tauri/src/commands/ai/commands.rs @@ -253,6 +253,42 @@ pub async fn ai_chat_clear(state: State<'_, AppState>) -> Result<(), String> { Ok(()) } +/// 强制发送消息（B-260616-02: L2 发送韧性） +/// +/// 当后端 generating=true 残留（HMR/异常退出等）导致 sendMessage 被拦截时， +/// 前端可调此命令强制复位 generating + 清审批，然后走 ai_chat_send 同款流程发消息。 +/// 等价于"先软停止 → 再发送"的原子操作，避免竞态窗口。 +#[tauri::command] +pub async fn ai_chat_force_send( + app: AppHandle, + state: State<'_, AppState>, + message: String, + language: Option, + skill: Option, +) -> Result { + // 原子复位:清 generating + 清积压审批 + 置 stop_flag,与 ai_chat_stop 审批分支一致 + let old_conv_id = { + let mut session = state.ai_session.lock().await; + let old = session.active_conversation_id.clone(); + session.generating = false; + session.pending_approvals.clear(); + session.stop_flag.store(true, Ordering::SeqCst); + old + }; + // 通知前端旧生成已结束(若有残留 conv) + if let Some(ref cid) = old_conv_id { + let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + total_tokens: 0, + prompt_tokens: 0, + completion_tokens: 0, + conversation_id: Some(cid.clone()), + }); + } + // 复位完成后走 ai_chat_send 同款流程（内部会重新设 generating=true 并 spawn loop） + // 直接内联而非递归调 ai_chat_send，避免 IPC 嵌套 + ai_chat_send(app, state, message, language, skill).await +} + /// 停止当前 AI 生成 /// /// 两种场景： diff --git a/src-tauri/src/commands/ai/prompt.rs b/src-tauri/src/commands/ai/prompt.rs index 818dbad..ff7a2ac 100644 --- a/src-tauri/src/commands/ai/prompt.rs +++ b/src-tauri/src/commands/ai/prompt.rs @@ -61,7 +61,9 @@ fn system_prompt_parts(lang: &str) -> (&'static str, &'static str) { You can perform the following actions via tool calls:\n\ - Create/query projects, tasks, and ideas\n\ - Run workflows\n\ - - Read file contents, list directories, create/write files\n\n\ + - Read file contents, list directories, create/write files\n\ + - Get file metadata (size, line count, binary detection)\n\ + - Append content to files, search files by name pattern\n\n\ ## Guidelines\n\ - Briefly explain your intent before executing actions\n\ - Ask for clarification if the user's intent is unclear\n\ @@ -76,7 +78,9 @@ fn system_prompt_parts(lang: &str) -> (&'static str, &'static str) { 你可以通过工具调用执行以下操作：\n\ - 创建/查询项目、任务、灵感\n\ - 运行工作流\n\ - - 读取文件内容、列出目录、创建/写入文件\n\n\ + - 读取文件内容、列出目录、创建/写入文件\n\ + - 获取文件元信息（大小、行数、二进制检测）\n\ + - 追加写入文件、按名称模式搜索文件\n\n\ ## 行为准则\n\ - 执行操作前简要说明你的意图\n\ - 如果不确定用户意图，先提问\n\ diff --git a/src/api/ai.ts b/src/api/ai.ts index d3feab3..5e3c22f 100644 --- a/src/api/ai.ts +++ b/src/api/ai.ts @@ -10,6 +10,11 @@ export const aiApi = { return invoke('ai_chat_send', { message, language: language || 'zh-CN', skill: skill || null }) }, + /** 强制发送（B-260616-02: 复位 generating 残留后走 send 同款流程） */ + forceSend(message: string, language?: string, skill?: string): Promise { + return invoke('ai_chat_force_send', { message, language: language || 'zh-CN', skill: skill || null }) + }, + /** 批准/拒绝工具调用 */ approve(toolCallId: string, approved: boolean): Promise { return invoke('ai_approve', { toolCallId, approved }) diff --git a/src/components/AiChat.vue b/src/components/AiChat.vue index 618932c..6b91989 100644 --- a/src/components/AiChat.vue +++ b/src/components/AiChat.vue @@ -2,7 +2,17 @@
-
+
+ +
{{ $t('aiChat.sidebarTitle') }}
@@ -98,6 +117,14 @@ {{ $t('ai.assistant') }} +
+ ⏳ {{ $t('aiChat.pendingApprovalCount', { n: pendingApprovalCount }) }} +
+
+ + {{ formatRelativeZh(msg.timestamp) }}
@@ -192,14 +252,28 @@
+
-
- + + + +
+
@@ -221,7 +295,15 @@ ref="toolCardListRef" :toolCalls="msg.toolCalls" @approve="({ id, approved }) => store.approveToolCall(id, approved)" + @batch-approve="(decision) => store.batchApprove(decision)" /> + + + {{ formatRelativeZh(msg.timestamp) }}
@@ -229,7 +311,7 @@ - @@ -238,11 +320,27 @@
+ +
+ + {{ agenticProgressText }} +
-
+
- {{ $t('aiChat.queueTitle', { n: store.state.queue.length }) }} - + {{ $t('aiChat.queueTitle', { n: store.state.queue.length }) }}({{ queueWaitSeconds }}s) +
+ + +
@@ -252,11 +350,14 @@
- +
- /{{ pendingSkill.name }} - {{ pendingSkill.description }} - + + /{{ pendingSkill.name }} + {{ pendingSkill.description }} + {{ pendingSkill.argument_hint }} + +