diff --git a/Cargo.lock b/Cargo.lock index 4bb4dca..10d4ef5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1039,7 +1039,7 @@ dependencies = [ [[package]] name = "devflow" -version = "0.1.1" +version = "0.1.2" dependencies = [ "anyhow", "async-trait", @@ -1059,7 +1059,10 @@ dependencies = [ "df-workflow", "futures", "futures-util", + "hex", + "hmac", "htmd", + "httpdate", "keyring", "larksuite-oapi-sdk-rs", "percent-encoding", @@ -1067,6 +1070,8 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "sha1 0.10.6", + "sha2 0.10.9", "sys-locale", "tauri", "tauri-build", @@ -1097,6 +1102,7 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "base64 0.22.1", "bytes", "df-ai-core", "df-types", @@ -1299,6 +1305,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", "crypto-common 0.1.7", + "subtle", ] [[package]] @@ -2173,6 +2180,15 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest 0.10.7", +] + [[package]] name = "htmd" version = "0.5.5" diff --git a/crates/df-ai-core/src/provider.rs b/crates/df-ai-core/src/provider.rs index 5d92f6e..a0c3a9e 100644 --- a/crates/df-ai-core/src/provider.rs +++ b/crates/df-ai-core/src/provider.rs @@ -42,25 +42,25 @@ impl ContentPart { impl ChatMessage { pub fn system(content: impl Into) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::System, content: content.into(), parts: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::System, content: content.into(), parts: None, source: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } pub fn user(content: impl Into) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::User, content: content.into(), parts: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::User, content: content.into(), parts: None, source: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } pub fn assistant(content: impl Into) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::Assistant, content: content.into(), parts: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::Assistant, content: content.into(), parts: None, source: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } pub fn assistant_with_tools(content: impl Into, tool_calls: Vec) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::Assistant, content: content.into(), parts: None, tool_call_id: None, tool_calls: Some(tool_calls), model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::Assistant, content: content.into(), parts: None, source: None, tool_call_id: None, tool_calls: Some(tool_calls), model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } pub fn tool_result(call_id: impl Into, content: impl Into) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::Tool, content: content.into(), parts: None, tool_call_id: Some(call_id.into()), tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::Tool, content: content.into(), source: None, parts: None, tool_call_id: Some(call_id.into()), tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } /// 多模态 user 消息：content 文本 + parts（含 Image 片）。 /// content 作为人类可读文本（也作非 vision 端点降级载荷）；parts 透传给 vision 端点。 pub fn user_parts(content: impl Into, parts: Vec) -> Self { - Self { id: Some(new_message_id()), role: MessageRole::User, content: content.into(), parts: Some(parts), tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } + Self { id: Some(new_message_id()), role: MessageRole::User, content: content.into(), parts: Some(parts), source: None, tool_call_id: None, tool_calls: None, model: None, status: None, reasoning_content: None, prompt_tokens: None, completion_tokens: None, prompt_cache_hit_tokens: None, prompt_cache_miss_tokens: None, reasoning_tokens: None, is_estimated: None, timestamp: Some(now_millis_i64()) } } /// 是否含图片片（供 provider 判定走多模态分支）。 diff --git a/crates/df-ai-core/src/types.rs b/crates/df-ai-core/src/types.rs index 99f96c9..69d9813 100644 --- a/crates/df-ai-core/src/types.rs +++ b/crates/df-ai-core/src/types.rs @@ -68,6 +68,22 @@ pub enum ContentPart { }, } +/// 消息来源(#958 溯源):外部渠道进入会话的原始消息定位。序列化为 JSON 随消息 +/// 落库(ai_messages.source 列);桌面本端消息恒 None 不落 +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct MessageSource { + /// 渠道标识:"feishu"(后续可扩 openapi/webhook 等) + pub channel: String, + /// 渠道侧原始消息 ID(飞书 om_/oc_ 前缀,可跳原消息追溯) + pub message_id: String, + /// 渠道侧会话/群 ID(可选,群聊定位) + #[serde(default, skip_serializing_if = "Option::is_none")] + pub chat_id: Option, + /// 发送者标识(飞书 open_id,可选) + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sender: Option, +} + /// 聊天消息 /// /// ⚠️ **迁移耦合点**:下方各字段名(`role`/`content`/`parts`/`tool_call_id`/ @@ -85,6 +101,11 @@ pub struct ChatMessage { pub id: Option, pub role: MessageRole, pub content: String, + /// 消息来源(#958 溯源):消息从哪个渠道/端点进入会话。None=桌面本端(既有语义, + /// 老数据反序列化为 None 零回归);Some=外部渠道(飞书 message_id 可跳原消息追溯, + /// 多端混发可区分来源端) + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, /// 多模态内容片。 /// /// `None`/空 → 纯文本消息（绝大多数场景，content 即全部载荷）。 diff --git a/crates/df-ai/Cargo.toml b/crates/df-ai/Cargo.toml index 6794041..48cb0ee 100644 --- a/crates/df-ai/Cargo.toml +++ b/crates/df-ai/Cargo.toml @@ -12,6 +12,7 @@ tokio = { workspace = true, features = ["sync", "time"] } async-trait = { workspace = true } anyhow = { workspace = true } tracing = { workspace = true } +base64 = "0.22" # HTTP + 流式 reqwest = { version = "0.12", features = ["stream", "json", "rustls-tls"] } diff --git a/crates/df-ai/src/anthropic_compat.rs b/crates/df-ai/src/anthropic_compat.rs index 40b6ca0..e88da43 100644 --- a/crates/df-ai/src/anthropic_compat.rs +++ b/crates/df-ai/src/anthropic_compat.rs @@ -288,9 +288,11 @@ impl AnthropicCompatProvider { /// 单个 ContentPart → Anthropic content block（text/image）。 /// - Text 片 → {type:text, text} /// - Image 片 → {type:image, source:{type:base64, media_type, data}} - /// base64 内嵌；url 模式当前不可达，兜底发空 data + warn（保留原行为）。 + /// base64 内嵌；本地 file:// 引用(#958)先展开为 base64，http(s) url 模式 + /// 当前不可达，兜底发空 data + warn（保留原行为）。 /// match 取得 p 所有权后直接 move media_type/base64，避免大 base64 clone。 fn content_part_to_block(p: crate::provider::ContentPart) -> serde_json::Value { + let p = crate::image_ref::inline_local_image_part(p); match p { crate::provider::ContentPart::Text { text } => serde_json::json!({ "type": "text", @@ -427,6 +429,22 @@ impl AnthropicCompatProvider { .and_then(|m| m.get("role").and_then(|r| r.as_str())) .unwrap_or(""); if first_role == "user" { + // 首条 user 但 content 为空串(会话压缩/续发残留,典型 text(0B)):原样发出会被 + // precheck "user content 空" 拦下而整请求失败。与上方补占位同思路——就地填 + // 占位文案(空串无信息可丢),上游实测接受非空占位,precheck 亦放行。 + let empty_str = messages[0] + .get("content") + .and_then(|c| c.as_str()) + .map(|s| s.is_empty()) + .unwrap_or(false); + if empty_str { + warn!( + msg_count = messages.len(), + "ensure_leading_user: 首条 user content 为空串,填占位(防 precheck 拦截)" + ); + messages[0]["content"] = + serde_json::json!("(continued from previous context)"); + } return; } // 空 Vec(异常会话经 sanitize 清空)或首条非 user → 补 user 占位: @@ -1163,6 +1181,46 @@ mod tests { assert_eq!(content[1]["source"]["data"], "iVBOR"); } + /// 本地 file:// 引用(#958)在转换层展开为 base64 source;缺失降级 text 块不炸请求 + #[test] + fn anthropic_convert_expands_local_file_ref() { + use base64::Engine as _; + let _convert = convert_lock(); + let dir = std::env::temp_dir().join("df-ai-anthropic-ref-test"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("ref.png"); + std::fs::write(&path, b"\x89PNG ok").unwrap(); + + let provider = AnthropicCompatProvider::new("https://api.anthropic.com", "k", "claude-3-5-sonnet"); + let req = CompletionRequest { + model: "claude-3-5-sonnet".into(), + messages: vec![ChatMessage::user_parts( + "看图", + vec![crate::provider::ContentPart::Image { + url: Some(format!("file://{}", path.display())), + base64: None, + media_type: Some("image/png".into()), + alt: None, + }], + )], + temperature: None, + max_tokens: None, + stream: false, + tools: None, + tool_choice: None, + reasoning_content: None, + }; + let body = provider.convert_request(req); + let user_msg = body + .messages + .iter() + .find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user")) + .expect("应有 user 消息"); + let content = user_msg.get("content").and_then(|c| c.as_array()).expect("user content 应为数组"); + assert_eq!(content[1]["source"]["data"], base64::engine::general_purpose::STANDARD.encode(b"\x89PNG ok")); + std::fs::remove_file(&path).ok(); + } + /// 纯文本 user 消息 → content 仍是字符串简写（无图不数组化） #[test] fn anthropic_convert_text_only_user_remains_string() { diff --git a/crates/df-ai/src/image_ref.rs b/crates/df-ai/src/image_ref.rs new file mode 100644 index 0000000..6cc8769 --- /dev/null +++ b/crates/df-ai/src/image_ref.rs @@ -0,0 +1,122 @@ +//! 本地图片引用展开(#958 图片入站引用化) +//! +//! 消息持久化只存引用(ContentPart::Image.url = `file://<绝对路径>`,base64 空), +//! 发送请求前在此统一展开为 base64 内嵌——存储面瘦 referencing(一张 5MB 图不再以 +//! 6.7MB base64 字符串进 SQLite 随历史全量重放),转换面按需读盘(SSD 毫秒级, +//! 对比 base64 内嵌的每请求全量携带,磁盘读便宜得多)。 +//! +//! 展开点是 provider 转换层的入口:主会话/子代理/标题/压缩等全部请求路径在此 +//! 单点覆盖,调用方(飞书入站/桌面粘贴图)只管落盘产引用,不关心 provider 差异。 +//! +//! 失败语义:引用文件丢失/不可读时降级为 Text 片「[图片文件缺失: path]」—— +//! 诚实告知模型图不可用,不让请求 400(空 base64 会被两个协议端点拒绝)。 + +use crate::provider::ContentPart; +use base64::Engine as _; + +/// `file://` 引用前缀。内部引用标识(两端自洽即可,不进端点):落盘方拼 +/// `file://<原生绝对路径>`,展开方 strip 后按原生路径读(Windows 盘符路径直接可用)。 +pub const LOCAL_FILE_SCHEME: &str = "file://"; + +/// 判定 Image 片是否为本地引用(base64 空 + url 带本地 scheme) +fn is_local_ref(url: Option<&str>, base64: Option<&str>) -> bool { + base64.is_none() && url.is_some_and(|u| u.starts_with(LOCAL_FILE_SCHEME)) +} + +/// 把消息内的本地图片引用展开为 base64 内嵌片;非本地片原样透传。 +/// 文件缺失/不可读时该 Image 片降级为 Text 片(请求不炸,模型得到可读的缺失说明)。 +pub fn inline_local_image_parts(parts: Vec) -> Vec { + parts.into_iter().map(inline_local_image_part).collect() +} + +/// 单片版(provider 转换臂逐片 map 用) +pub fn inline_local_image_part(p: ContentPart) -> ContentPart { + match p { + ContentPart::Image { url, base64, media_type, alt } + if is_local_ref(url.as_deref(), base64.as_deref()) => + { + let path = url.unwrap(); + match std::fs::read(&path[LOCAL_FILE_SCHEME.len()..]) { + Ok(bytes) => ContentPart::Image { + url: None, + base64: Some(base64::engine::general_purpose::STANDARD.encode(bytes)), + media_type, + alt, + }, + Err(_) => ContentPart::Text { + text: format!("[图片文件缺失: {}]", &path[LOCAL_FILE_SCHEME.len()..]), + }, + } + } + other => other, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn local_ref_expands_to_base64() { + let dir = std::env::temp_dir().join("df-ai-image-ref-test"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("expands.png"); + std::fs::write(&path, b"\x89PNG fake-bytes").unwrap(); + + let out = inline_local_image_parts(vec![ContentPart::Image { + url: Some(format!("file://{}", path.display())), + base64: None, + media_type: Some("image/png".into()), + alt: None, + }]); + match &out[0] { + ContentPart::Image { url, base64, media_type, .. } => { + assert!(url.is_none()); + let b = base64.as_deref().expect("base64 应已填充"); + assert_eq!(b, base64::engine::general_purpose::STANDARD.encode(b"\x89PNG fake-bytes")); + assert_eq!(media_type.as_deref(), Some("image/png")); + } + other => panic!("应保持 Image 片, 实际 {other:?}"), + } + std::fs::remove_file(&path).ok(); + } + + #[test] + fn missing_file_degrades_to_text_not_empty_base64() { + let out = inline_local_image_parts(vec![ContentPart::Image { + url: Some("file:///nonexistent/df-ai-no-such.png".into()), + base64: None, + media_type: Some("image/png".into()), + alt: None, + }]); + match &out[0] { + ContentPart::Text { text } => { + assert!(text.contains("[图片文件缺失"), "应降级为缺失说明, 实际 {text}"); + assert!(text.contains("df-ai-no-such.png")); + } + other => panic!("缺失文件应降级 Text 片, 实际 {other:?}"), + } + } + + #[test] + fn non_local_parts_pass_through() { + let passthrough = vec![ + ContentPart::text("hello"), + ContentPart::image_base64("image/png", "aGVsbG8="), + ContentPart::image_url("https://x/a.png"), + // base64 非空的 file:// 不是引用(优先内嵌),原样透传 + ContentPart::Image { + url: Some("file:///x.png".into()), + base64: Some("aGVsbG8=".into()), + media_type: Some("image/png".into()), + alt: None, + }, + ]; + let out = inline_local_image_parts(passthrough); + assert_eq!(out.len(), 4); + assert!(matches!(&out[0], ContentPart::Text { .. })); + assert!(matches!(&out[1], ContentPart::Image { base64: Some(_), .. })); + assert!(matches!(&out[2], ContentPart::Image { url: Some(u), .. } if u.starts_with("https://"))); + assert!(matches!(&out[3], ContentPart::Image { base64: Some(_), .. })); + } +} diff --git a/crates/df-ai/src/lib.rs b/crates/df-ai/src/lib.rs index 6882b9b..7ac0588 100644 --- a/crates/df-ai/src/lib.rs +++ b/crates/df-ai/src/lib.rs @@ -33,6 +33,8 @@ mod model_probe_helpers; pub mod plan_hint; pub mod openai_compat; pub mod openai_helpers; +// 本地图片引用展开(#958 图片入站引用化):消息存 file:// 引用,发送前单点展开 base64 +pub mod image_ref; // Phase 0 (B 路线): Plan-driven 规划骨架(维度 0.8 Plan/SubTask 结构 + 0.7 规划校验/Kahn 分层)。 // 纯函数模块(零 IO 零状态)。依据 docs/02-架构设计/单对话并行多轮-{设计,Phase0落地路线图}-2026-06-20.md。 // Phase 1(plan_hint 接入主 loop)/Phase 2(planning)/Phase 3(并行 execution) 接入后续。 @@ -50,6 +52,9 @@ pub mod router; pub mod retry; pub mod sse_parser; pub mod namespace_store; +// P1 工具结果瘦身(上下文根治专项 2026-10-06 设计 §二):超阈值工具结果正文落盘 + 句柄。 +// 与 src-tauri 审计侧 D2a(tool_result_offload) 同构;base 未注入时降级为纯摘要(零行为变化)。 +pub mod tool_offload; use provider::LlmProvider; use reqwest::Client; diff --git a/crates/df-ai/src/openai_compat.rs b/crates/df-ai/src/openai_compat.rs index 1f242b4..656d052 100644 --- a/crates/df-ai/src/openai_compat.rs +++ b/crates/df-ai/src/openai_compat.rs @@ -118,10 +118,12 @@ impl OpenAICompatProvider { // 多模态 content（须在 move m.tool_calls 之前算，借用 m）。 // 含图消息走 content 数组（text/image_url）；纯文本走字符串简写 // （保持与现有纯文本端点零回归）。image_url 支持 data URI（base64）与 http(s) URL。 + // 本地 file:// 引用(#958)先展开为 base64 内嵌——OpenAI 端点不认本地路径 let content = if m.has_image() { let parts: Vec = m .flattened_parts() .into_iter() + .map(crate::image_ref::inline_local_image_part) .map(|p| match p { crate::provider::ContentPart::Text { text } => serde_json::json!({ "type": "text", @@ -1020,6 +1022,57 @@ mod tests { assert_eq!(arr[2]["image_url"]["url"], "https://x/a.png"); } + /// 本地 file:// 引用(#958)在转换层展开为 data URI;引用缺失降级 Text 片不炸请求 + #[test] + fn openai_convert_expands_local_file_ref() { + use base64::Engine as _; + let dir = std::env::temp_dir().join("df-ai-openai-ref-test"); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("ref.png"); + std::fs::write(&path, b"\x89PNG ok").unwrap(); + + let provider = OpenAICompatProvider::new("https://api.openai.com", "k", "gpt-4o"); + let req = CompletionRequest { + model: "gpt-4o".into(), + messages: vec![ + ChatMessage::user_parts( + "看图", + vec![crate::provider::ContentPart::Image { + url: Some(format!("file://{}", path.display())), + base64: None, + media_type: Some("image/png".into()), + alt: None, + }], + ), + ChatMessage::user_parts( + "坏引用", + vec![crate::provider::ContentPart::Image { + url: Some("file:///nonexistent/ref-missing.png".into()), + base64: None, + media_type: Some("image/png".into()), + alt: None, + }], + ), + ], + temperature: None, + max_tokens: None, + stream: false, + tools: None, + tool_choice: None, + reasoning_content: None, + }; + let out = provider.convert_request(req); + let good = &out.messages[0]; + let arr = good.content.as_array().expect("含图 → 数组"); + assert_eq!(arr[1]["image_url"]["url"], format!("data:image/png;base64,{}", base64::engine::general_purpose::STANDARD.encode(b"\x89PNG ok"))); + + let bad = &out.messages[1]; + let arr = bad.content.as_array().expect("降级片仍在数组"); + assert_eq!(arr[1]["type"], "text"); + assert!(arr[1]["text"].as_str().unwrap().contains("[图片文件缺失")); + std::fs::remove_file(&path).ok(); + } + /// 纯文本消息 → content 仍是字符串简写（无图不数组化，对齐纯文本端点兼容） #[test] fn openai_convert_text_only_remains_string() { diff --git a/crates/df-storage/src/crud/message_repo.rs b/crates/df-storage/src/crud/message_repo.rs index cad232b..8eb7dde 100644 --- a/crates/df-storage/src/crud/message_repo.rs +++ b/crates/df-storage/src/crud/message_repo.rs @@ -47,6 +47,7 @@ fn ai_message_from_row(row: &Row<'_>) -> std::result::Result Result<()> { // conversation_checkpoints.token_total 语义回填。 // V51 = 待发箱 ai_outbox 表(定时消息与待发箱持久化 P0 地基):队列/定时统一实体, // 调度器按 status+due_at 索引到点派发。纯新增表,可逆=DROP TABLE ai_outbox。 - let steps: [(i32, fn(&Connection) -> Result<()>); 51] = [ + // V52 = ai_messages.source 列(#958 消息溯源:外部渠道原始消息 ID JSON,NULL=桌面本端)。 + let steps: [(i32, fn(&Connection) -> Result<()>); 52] = [ (1, migrate_v1), (2, migrate_v2), (3, migrate_v3), @@ -122,6 +123,8 @@ pub fn run(conn: &Connection) -> Result<()> { (50, migrate_v50), // V51 = 待发箱 ai_outbox 表(定时消息与待发箱持久化 P0 地基)。 (51, migrate_v51), + // V52 = ai_messages 加 source 列(#958 消息溯源:飞书等外部渠道原始消息 ID)。 + (52, migrate_v52), ]; for (version, migrate_fn) in steps { @@ -1529,7 +1532,7 @@ fn migrate_v50(conn: &Connection) -> Result<()> { /// 调度器启动时 reset_stale_sending 复位为 pending。 /// 迁移链顶点版本(单一真值来源)：**新增迁移时必须同步 +1**,否则下方迁移测试会失败 /// (测试用它断言「全量迁移后 schema_version == 顶点」「重跑仍到顶点」——既防链中断,也免多处写死)。 -pub const LATEST_SCHEMA_VERSION: i32 = 51; +pub const LATEST_SCHEMA_VERSION: i32 = 52; fn migrate_v51(conn: &Connection) -> Result<()> { if !table_exists(conn, "ai_outbox") { @@ -1569,6 +1572,18 @@ fn migrate_v51(conn: &Connection) -> Result<()> { Ok(()) } +/// V52: ai_messages 加 source 列(#958 消息溯源)。TEXT JSON(MessageSource 序列化), +/// NULL=桌面本端。纯加列零回退风险(老数据恒 NULL=既有语义)。 +fn migrate_v52(conn: &Connection) -> Result<()> { + if !column_exists(conn, "ai_messages", "source") { + conn.execute("ALTER TABLE ai_messages ADD COLUMN source TEXT", [])?; + tracing::info!("v52: ai_messages 加 source 列(消息溯源)"); + } + conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [52])?; + tracing::info!("迁移 v52 完成"); + Ok(()) +} + /// 回填 conversation_checkpoints.token_total:按 snapshot JSON 内逐条消息的 /// prompt_tokens+completion_tokens 求真实总量。 /// diff --git a/crates/df-storage/src/models.rs b/crates/df-storage/src/models.rs index abc8218..8248375 100644 --- a/crates/df-storage/src/models.rs +++ b/crates/df-storage/src/models.rs @@ -479,6 +479,9 @@ pub struct AiMessageRecord { /// 本轮 prompt 是否估算值(round_usage.prompt_tokens==0 → estimated_prompt 兜底打标)。 /// reload 逐条回显「估算」标注;老消息 NULL → None(向前兼容)。 pub is_estimated: Option, + /// 消息来源 JSON(#958 溯源,MessageSource 序列化;V52 加列)。 + /// NULL=桌面本端(既有语义);非空=外部渠道(飞书 message_id 等,可跳原消息追溯) + pub source: Option, } /// 待发箱记录(ai_outbox 表,V51)。 diff --git a/docs/todo.md b/docs/todo.md index ea2d850..6343170 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -716,3 +716,29 @@ - 拍板点：① 是否引入「非终态 tool_result」这一语义（默认关可回退）；② grace 秒默认值；③ 后台期间会话态展示（Generating 还是「后台执行中」）。 937. [ ] **对话内创建定时/延时任务(2026-10-10 用户报)**:在对话中直接创建「定时执行/延时执行」的任务(如「1 小时后跑 X」)。**需设计**:与既有 Tasks/工作流(advance_task、run_workflow)的关系、调度落点(与 935 共用调度器)、无人值守引擎(unattended)是否复用、UI 呈现(任务卡带定时徽标)。设计落 docs 后再实施。 + +943. [ ] **ai-chat UI 检查:待发送队列横幅(2026-10-10 用户截图实证+代码核验)**——位置 `AiChat.vue:79-113`(样式 965-1050): + - **UI-1 全文不可辨**(中):`.ai-queue-text`(1005-1011)单行 ellipsis 且**无 title** → 长路径+指令完全无法看全;修:补 `:title="item.text"` 或两行 clamp。 + - **UI-2 视觉层级倒挂**(低):横幅 accent 底色+多操作,比主输入区还醒目;待发是次要动作,建议降噪(中性底色/收起为一条计数)。 + - **UI-3 操作语义重叠**(低):「回填(108)」与「编辑(110)」并存——回填=取出放回输入框,编辑=行内编辑,功能高度重叠;建议二合一。 + - **UI-4 清空无确认**(中):`clearQueue`(84)直接清空全部待发,**无二次确认**;对比「强制发送」有 confirm(`tryForceSend`,533);误点即丢 10 条。修:加 confirmDialog。 + - **UI-5 无上下文预期**(低):有等待时长(queueItemWait ✓)但无「预计还要等多久/当前生成还剩多少」;信息可后置,仅登记。 + +945. [ ] **ai-chat UI 检查:工具卡头部截断(2026-10-10 截图实证)**——`ToolCard.vue:27/:367`:头部摘要 `resultSummary` 有 `:title` 悬停,但 `summaryTitle` **只取 `args.path`**——`run_command` 类悬停看不到命令内容,头部命令文本又单行截断(446-450) → 截图实证「$ …命令成功 (19s)」完全不可辨。修:`summaryTitle` 按工具类型取关键参数(run_command→command;无 path 工具→首参)。 + +947. [ ] **ai-chat UI 检查:token 统计常驻每条消息(2026-10-10 截图实证)**——每条带 reasoning 的消息都挂思考 pill,右端 token 小字(`roundTokenText`,MessageList)逐条重复「55.4k in(估算) 974 c…」且截断显示不全 → 元信息噪音化。修(方向,需拍板):① pill 悬停才显 token;② 仅最新一显,历史轮收进轨迹卡统计;③ 截断点保底(最少显全「974 out」段)。 + +949. [ ] **强制发送后发送按钮失态(2026-10-10 用户实测报)**:队列「强制发送」后,后端在新 loop 正常生成(内容照常流出),但发送按钮回到普通「发送」态而非「停止」态。**根因方向(已定位待实施)**:force_send 复位旧 generating 时 emit 旧任务 AiCompleted(`chat.rs` 复位分支)——前端迟到收尾仲裁(useAiLifecycleEvents:60,IPC 查后端 generating)存在 ① 事件到达 vs 新 loop 占用 generating 的竞态窗口(查到 false 即误判真收尾);② AiTextDelta 无条件累加所以内容照流、streaming 却被翻 false,内容与按钮态脱钩正好吻合现象。修(三选一或组合,实施时细查 emit 顺序):① 后端把旧终态 emit 挪到新 loop 占用 generating 之后(消竞态);② 前端 AiConvStateChanged/相关收敛路径全纳入仲裁;③ 自愈兜底:AiHeartbeat 到达时若后端 generating=true 而前端 streaming=false → 置回 true。验收:强制发送后按钮全程「停止」态直至收尾。 +951. [ ] **消息聚合形式优化(2026-10-10 用户提出)**:进一步聚合消息展示粒度——当前「结果/过程分离」是一桶一张轨迹卡(每轮一张);长任务(截图 20 轮)仍会有 N 张卡。方向(待拍板):① 任务级聚合=一次发送的全程收敛为**单张过程卡**(内部分轮导航/折叠),气泡只留最终结果;② 维持每轮一卡但加强轮间视觉折叠(连续轮默认收进「+N 轮」);③ 先上已实现的「结果/过程分离」新版实测再定(可能已够)。依赖:新版换装后实测再拍板粒度。 + +953. [ ] **主对话 max_tokens 截断无 UI 提示 + 续写整段重发(2026-10-10 用户新版实测截图,魔芋渠道接入会话)**:模型输出 921 out 戛然而止(finish_reason=length),气泡**无任何截断标记**——用户只能问「卡了吗」;追问后模型答「上一轮被截断,接着整理完整」却**从头发一遍**,同一方案在列表里重复两大段(token 翻倍+滚动负担)。代码核验:后端有专门截断标志(stream_recv.rs:105 truncated_by_max_tokens,openai_helpers.rs:341 finish_reason=length),但仅子代理完成卡消费(AgentFinishedStatus truncated);主对话气泡零消费。修(三层):①**UI 提示**:truncated_by_max_tokens 穿透到主对话气泡——尾部渲染「⚠ 输出因长度上限截断」弱提示条(附「继续输出」按钮=注入续写指令,复用 16 纠偏注入通道);②**行为**:续写注入让模型从断点续(上下文半截已在,提示「从中断处继续,勿重复已输出内容」),防整段重发;③**配置面**:921 out 即断疑似该渠道 max_tokens 上限小或默认 max_tokens 偏小,查 model_configs 该渠道 max_tokens 配置。验收:截断气泡有可见提示;继续输出不重复已发内容。 + - **③ 配置面结论(2026-10-10 核验)**:`model_configs` 全部 `max_tokens=None`(ds-flash/glm/moyu 系 8 条)→ 请求不带输出上限,断点取决于上游默认;921 out 即断=该渠道模型输出上限小,且当前配置结构**无法干预**(model_configs 不支持 max_tokens 字段)→ 能力缺口:给 ModelConfig 增 max_tokens 可选字段+请求构造透传(涉 crates/df-ai,待解禁)。另:`agentic/mod.rs:2271` 截断后已有「以上内容可能不完整;可要求继续输出」system 提示(**只给模型,UI 不可见**)——模型能答「上轮被截断」正源于此,953① 的用户可见提示仍必要。旁证:DB 中 ctx=8192 哨兵大量存在,用户实测 56.7k in 未被卡 → J2 哨兵不采信守卫生效 ✓。 + +955. [ ] **DSML 工具调用标记泄漏为可见文本(2026-10-10 用户实测,魔芋渠道 ds-flash)**:模型输出 `<｜｜DSML｜｜ invoke name="run_command">`、`<｜｜DSML｜｜ parameter name="command" string="true">cd E:\wk-lab\devflow; npx vue-tsc...` 等 DeepSeek 文本协议工具调用标记,整段泄漏为可见文本(工具卡不渲染、命令裸露)。**根因(全仓 grep 取证)**:Rust/前端**均无 DSML 解析代码**——DevFlow 流解析只认原生 `tool_calls` delta;魔芋渠道的 DeepSeek v3.1+ 在该渠道不走原生 function calling 而是把工具调用以 DSML 特殊标记写进 content 文本(上游行为),无人解析即泄漏;且截断(921 out)发生在标记中间时,残缺 DSML 必然泄漏。**修(三层)**:①流解析层:openai_helpers/stream_recv 检测 content 中的 DSML invoke/parameter 标记 → 解析出 name/parameters 构造 ToolCallDraft(与原生 tool_calls 同路,工具卡正常渲染+真实执行);②渲染兜底:MessageItem/Markdown 渲染前剥离未闭合 DSML 残段(防截断泄漏);③**模型/渠道规避**:该行为是 DeepSeek v3.1+ 渠道侧特性,魔宇渠道换原生 function-calling 模型或升级渠道适配可根治。验收:DSML 工具调用正常出卡并执行;截断残段不泄漏。**注意**:全仓引入 DSML 解析是新能力,实施前按「新能力引入评估纪律」走设计档。 +957. [ ] **审计决策者键补齐(2026-10-10 Wave G 发现)**:审计决策者列后端还有 5 个值渲染裸 key:`auto_retry_guard`(audit/approval.rs:185)、`auto_resolve_fail`(audit/mod.rs:750)、`boundary-dirs`(audit/mod.rs:909)、`boundary_auto`(audit/mod.rs:1449/1517)、`sub_agent_denied`(audit/mod.rs:116)——auditLog.ts 的 decided 段只有 8 键,缺这 5 个。修:i18n zh/en 补 5 键(值按语义直译);其中 boundary-dirs/sub_agent_denied 语义为拒绝,不要回退到既有 auto 键。 + +958. [x] **飞书通道图片入站能力(2026-10-11 用户发图实测,回「暂仅支持文本」提示卡)**:用户给 DevFlow 飞书 bot 发图,只收到降级提示——**勘察结论(AI-Ops 10-11,链路+可行性已核)**:非故障,系有意兜底(feishu_channel.rs:387,#25 静默丢修复);桌面端 ai-chat 图片链路**已完整**(ChatInput 粘贴/上传 → ContentPart::Image → chat.rs:920 has_image 自动转 vision 端点,anthropic 侧 url→base64 预拉已有),缺口**仅在飞书入站**。修(四件):①`feishu_channel.rs::handle_message` 增 image 分支(现 `message_type != "text"` 一刀切回提示卡);②拉图:飞书 im/v1/messages/{message_id}/resources/{file_key}?type=image 下载(file_key 从 image 消息 content JSON 取),SDK 无现成方法则手写 GET(租户 access_token 复用既有 client);③组装:字节→base64→ContentPart::Image+可选 caption 文本拼 user 消息,走既有 spawn 会话链路(vision 路由零改动);④开放平台侧:应用补 im:resource 只读权限。**注意**:①权限申领是对外变更(飞书后台),先拍板;②大图限流(base64 膨胀,建议 >10MB 拒收提示);③文件/PDF 等其他媒体不在本条范围。验收:发图给 bot,AI 能看到图并按图回答;非图媒体仍回现有提示卡。**【10-11 AI-Ops 实施完成(用户令「直接推进」),待换装+权限开通】**:①②③已全部落地 feishu_channel.rs 单文件——类型三分流(text/image/其余提示卡,image_key 取自 SDK 归一化 msg.resources 零手解析)/`download_feishu_image`(走 channel_messaging().download_message_resource 便捷方法,与 reply/send 同风格)/魔数判 media_type(sniff_image_mime 四格式,未知拒收防端点 400)+10MB 限流/引导词 IMAGE_INBOUND_PROMPT 喂模型(飞书 image 独立类型无法带文字)/白名单后下载(未授权不烧流量)+占位发票前失败早退/注入旁路排除图片(inject 只收文本,诚实落受理守卫)。验证:cargo check -p devflow 0 error + feishu_channel 测试 23 passed(含新 image_mime_sniff_four_formats)。**待两件**:①换装生效(重启应用,等令);②飞书开放平台给应用补「获取与上传图片或文件资源」权限(im:message.resource:readonly,后台操作,未开通时下载失败会回可读错误卡)。**【10-11 追加:引用化存储改造(用户指令「下载或转文件存储引用,信息存储只存一个引用,模型使用按需转换」)】**:base64 内嵌版升级为落盘引用版——图片字节落盘 `%APPDATA%/top.1216.devflow/media//.`(save_media_file,按会话分目录,不主动 GC),消息只存 `file://` 引用(ContentPart.url,base64 空),发送请求时 df-ai 新模块 image_ref.rs 单点展开(`inline_local_image_part`:读盘→base64;缺失降级 Text「[图片文件缺失]」不炸请求)——openai_compat(map 链插入)+anthropic_compat(content_part_to_block 入口)两臂接入,provider 转换层单点覆盖全部请求路径(主会话/子代理/标题/压缩+历史重放),零契约变更(url 字段语义扩展)。桌面粘贴图 base64 内嵌同族膨胀,后续批接同一机制(登记待办)。验证:df-ai 全量 **578 passed 0 failed**(新增 image_ref 3 测+openai/anthropic file:// 端到端各 1)+devflow check 0 error+feishu_channel 23 passed。**【10-11 第二轮:媒体域全面重构完成(用户令「推」+补1-补7),全链落地】**:①media_store.rs 新模块(媒体域单一归属,防臃肿)——hash 去重落盘(sha256 前 16hex,同内容同文件)+阿里云 OSS 双写(PUT Header 签名手写,失败降级 file:// 不阻塞)+文件名=`[_<清洗原始名>]`(u-claw safeAttachName 模式:hash 保去重+原始名保可读,路径穿越清洗);②下载防线(对齐 u-claw):3 次指数退避+120s 超时+半写残件不留盘;③spawn offload 统一收口(send+resend 两处,vision 判定后):桌面粘贴图 base64→落盘引用,SQLite 不进大 base64;④消息溯源 V52(ai_messages.source 列,MessageSource{channel,message_id,chat_id,sender}),spawn 第 11 参穿透,飞书传真实 msg_id(可跳原消息追溯,多端混发可区分);⑤出站 [交付] 回传(u-claw replyLocalFiles 同模式):报告写 [交付] 路径行→自动上传飞书(图片→图片消息/其余→文件消息)+敏感名拦截(.env/id_rsa/token 等 8 特征);⑥引用统一走 url 字段(https=OSS 跨设备可解析/file://=本机 image_ref 展开),不自造 scheme。验证:cargo check --workspace 0 error+df-storage 214+12 绿/df-ai 578 绿/devflow 832 绿(新增 media_store 4+delivery 3+V52 迁移)。**待**:换装生效+飞书后台 im:resource 权限+media-oss-config KV 手配(设置面板后续批)。 + +959. [ ] **占用期消息必达·收件箱续投(2026-10-11 u-claw 对照,补6)**:用户在 bot 任务占用期发的消息,devflow 现状=QUEUE_BUSY_MSG 拒绝+用户须重发;u-claw 已进化为 pending 收件箱落盘+scanner 自动续投+卡片原地刷新「已进入待投队列(无需重发)」——重试责任不在用户。修:受理守卫拒绝路径改落盘收件箱(ai_outbox 表 V51 已有,复用)+原地 PATCH 提示;空闲自动续投+结果回推。验收:任务占用期发消息不丢,任务完成自动处理并回推结果。 +960. [ ] **连续消息聚合防抖「已收 N 条无补充即处理」(2026-10-11 u-claw 对照,补6)**:u-claw 连发消息聚合「📥 已收 %d 条,%ds 内无补充即开始处理」防抖后再统一处理;devflow 每条立即受理(连续消息第二条起被占用守卫拒)。修:飞书入站短窗防抖(聚合期间到达的同 chat 消息合并为一次任务,文本拼接+附件清单合并),配合 959 收件箱。设计先行(与队列/注入语义联动,防双头)。 +961. [ ] **桌面轮次交互定位重定义(2026-10-11 用户拍板,补7 三条)**:①轨迹卡片收敛:除主回复气泡外交互收敛到一张「有动态+可折叠」的轨迹卡(现有轮次折叠类型),**修缺陷:正生成时卡片展开不了**;②轮次语义纠正:现「压缩后即新轮次」定位错——正确定义=**一次信息发起=一个任务轮次,中途补充信息(inject)仍属同一轮,全部完成后新发起才是新轮**;③轮次编号稳定化:现编号不稳定(刚是 5 一会下面又出现 1)——应固定为「该对话的第 1 轮/第 2 轮」持久编号(与消息 seq/轮边界锚点绑定,不随视图压缩/折叠漂移)。涉及:MessageList 折叠模型+轮次归属模型(压缩边界 vs 发起边界)+编号生成规则。设计先行,拆批实施。 diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 4b79f0f..64a7112 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -82,6 +82,12 @@ futures-util = "0.3" # 使用 rustls-tls(非 native-tls),避免 Windows SChannel 同步阻塞 tokio 工作线程 # 致 BUG-2026-07-17(aichat 流式调用永久 hang,外层 tokio timeout 亦无法推进计时器)。 reqwest = { version = "0.12", default-features = false, features = ["json", "gzip", "brotli", "rustls-tls", "stream"] } +# media_store(#958 引用化):hash 去重文件名(sha2+hex)+ 阿里云 OSS PUT 上传签名(hmac+sha1) +sha2 = "0.10" +hex = "0.4" +hmac = "0.12" +sha1 = "0.10" +httpdate = "1" # fetch_url AI 工具:URL → markdown 文档嗅探(GET HTML → htmd 转 markdown → 去噪音 + 截断)。 # 替代 http_request 拿原始 HTML(噪声大、爆 token)。turndown.js 移植,只读 GET,与 http_request 共享 SSRF 防护。 htmd = "0.5" diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 13a090b..78f126a 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -695,6 +695,7 @@ pub async fn ai_chat_send( parts, mention_spans, true, + None, // #958:标题/内部路径无外部来源 ) .await .map(|_| "ok".to_string()) @@ -772,6 +773,8 @@ async fn reply_status_summary( prompt_cache_miss_tokens: 0, reasoning_tokens: 0, is_estimated: false, + // 零 token 收敛信号,非截断(看板 953;枚举新字段,Rust 字面量强制显式赋值)。 + truncated_by_max_tokens: false, incomplete: None, incomplete_reason: None, conversation_id: Some(target), @@ -804,6 +807,8 @@ pub(crate) async fn spawn_agentic_session( parts: Option>, mention_spans: Option>, pin_active: bool, + // #958 溯源:外部渠道(飞书等)原始消息定位,None=桌面本端。随 user 消息落库 + source: Option, ) -> Result { // 获取活跃提供商（只读，失败可直接返回，不影响生成标志） let provider_config = super::super::prompt::get_active_provider(state).await?; @@ -820,6 +825,19 @@ pub(crate) async fn spawn_agentic_session( other => (other, false), }; + // offload(#958 引用化,vision 判定后):存活图片片 base64 → media_store 落盘引用 + // (hash 去重,OSS 配置在则 https 引用跨设备可解析)。桌面粘贴图/前端上传在此统一收口 + // ——消息只存引用,SQLite 不进大 base64,历史重放零负担。落盘失败保 base64 片不丢图。 + // 在锁前做(与上方预拉同窗口,不持锁 I/O) + let parts = match parts { + Some(ps) if !ps.is_empty() => { + let oss_cfg = crate::media_store::read_oss_config(&state.db).await; + let store = crate::media_store::MediaStore::new(state.data_dir.clone(), oss_cfg); + Some(crate::media_store::offload_image_parts(ps, &store).await) + } + other => other, + }; + // 纯发图(无文本无技能)且模型非视觉:降级后无任何有效内容,原「[图片已忽略…]」占位串 // 会被当 user 消息喂模型(回复跑偏且浪费一轮 LLM 调用)。此处直接拒绝发送,错误经前端 // 发送失败通道透出;图文混合(有文本/技能)维持占位后缀现状。置于锁与消息 push 之前, @@ -917,9 +935,13 @@ pub(crate) async fn spawn_agentic_session( // F-260614-05 Phase 2c: parts 非空走 user_parts(多模态),空/None 走 user()(纯文本零回归)。 // user_parts 内部把 parts 挂到 ChatMessage.parts,provider 在 has_image() 为真时转 vision 端点。 if let Some(ps) = parts.as_ref().filter(|p| !p.is_empty()) { - conv.messages.push(ChatMessage::user_parts(&user_content, ps.clone())); + let mut m = ChatMessage::user_parts(&user_content, ps.clone()); + m.source = source.clone(); + conv.messages.push(m); } else { - conv.messages.push(ChatMessage::user(&user_content)); + let mut m = ChatMessage::user(&user_content); + m.source = source.clone(); + conv.messages.push(m); } // 新用户轮把本请求钉扎为 WorkingContext 常驻目标(截断 200 字符,与 agentic G1 目标钉扎 GOAL_MAX_CHARS 同量级)。 // 喂原文 message 而非落库变换文本 user_content:begin_turn 契约是"本条用户请求", @@ -1105,6 +1127,7 @@ pub(crate) fn orchestration_session_spawner() -> df_nodes::ai_agent_node::SpawnF None, None, false, + None, // #958:编排路径无外部来源 ) .await { @@ -2267,6 +2290,16 @@ pub async fn ai_chat_force_send( other => (other, false), }; + // offload(#958 引用化,同 ai_chat_send):重发路径同样收口 base64→引用 + let parts = match parts { + Some(ps) if !ps.is_empty() => { + let oss_cfg = crate::media_store::read_oss_config(&state.db).await; + let store = crate::media_store::MediaStore::new(state.data_dir.clone(), oss_cfg); + Some(crate::media_store::offload_image_parts(ps, &store).await) + } + other => other, + }; + // 纯发图(无文本无技能)且模型非视觉:同 spawn_agentic_session 的拒绝闸(占位串不喂模型), // 置于锁与消息 push 之前,拒绝零副作用。 if degraded && message.trim().is_empty() && skill.is_none() { @@ -2400,25 +2433,6 @@ pub async fn ai_chat_force_send( session.conv(&conv_id).stop_flag.store(false, std::sync::atomic::Ordering::SeqCst); } - // 通知前端目标 conv 旧生成已结束(若先前在生成;emit 在锁外,避免持锁调 runtime emit) - if let Some(ref cid) = old_conv_id { - let ev = AiChatEvent::AiCompleted { - total_tokens: 0, - prompt_tokens: 0, - completion_tokens: 0, - prompt_cache_hit_tokens: 0, - prompt_cache_miss_tokens: 0, - reasoning_tokens: 0, - // 零 token 收敛信号,非估算(无真实 LLM 调用)。 - is_estimated: false, - incomplete: None, - incomplete_reason: None, - conversation_id: Some(cid.clone()), - pinned_goals: old_pinned_goals.clone(), - }; - emit_ai_chat_event(&app, ev); - } - // system prompt 构建(照 ai_chat_send):augmentation 注入 + 知识注入顺序一致。 let _tool_defs = state.ai_tools.tool_definitions(); let lang = language.unwrap_or_else(|| "zh-CN".to_string()); @@ -2478,9 +2492,50 @@ pub async fn ai_chat_force_send( sorted.first().map(|p| p.to_string_lossy().to_string()) }; + // 看板 949 后端消竞态:旧任务终态 emit 挪到「新 loop 占用 generating 之后」。 + // 原 emit 位于「复位+占用」与 spawn 之间:前端迟到收尾仲裁(useAiLifecycleEvents + // cleanupTerminatedConversation)收到旧 AiCompleted 时查 ai_is_generating,若新 loop 尚未 + // spawn / 入口未迁移 Generating 会读到 false → 误判真收尾 → setStreaming(false),新 loop + // 内容照流(AiTextDelta 无条件累加)但发送按钮失态。此处 spawn 后短轮询等新 loop 入口 + // GeneratingGuard 迁移 Generating(同步迁移,正常毫秒级),占用成功或 2s 到期后再 emit; + // 到期未占用也 emit(旧任务确已终局,前端按真收尾收敛语义仍正确,只是失去仲裁保护)。 + let occupied_probe = conv_id.clone(); tauri::async_runtime::spawn(async move { run_agentic_loop(session_arc, tools_arc, db, app_handle, provider_config, system_prompt, volatile_tail, conv_id, knowledge_config, llm_concurrency, max_iterations, max_retries, 0, conv_model_override, new_epoch, h3_watch_dir).await; }); + let occupied = { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + if state.conv_states.is_active(&occupied_probe) { + break true; + } + if std::time::Instant::now() >= deadline { + break false; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }; + tracing::debug!(conv_id = %occupied_probe, occupied, "[ai] force_send 旧终态 emit 前新 loop 占用检查"); + // 通知前端目标 conv 旧生成已结束(若先前在生成;emit 在锁外,避免持锁调 runtime emit) + if let Some(ref cid) = old_conv_id { + let ev = AiChatEvent::AiCompleted { + total_tokens: 0, + prompt_tokens: 0, + completion_tokens: 0, + prompt_cache_hit_tokens: 0, + prompt_cache_miss_tokens: 0, + reasoning_tokens: 0, + // 零 token 收敛信号,非估算(无真实 LLM 调用)。 + is_estimated: false, + // 零 token 收敛信号,非截断(看板 953;枚举新字段,Rust 字面量强制显式赋值)。 + truncated_by_max_tokens: false, + incomplete: None, + incomplete_reason: None, + conversation_id: Some(cid.clone()), + pinned_goals: old_pinned_goals.clone(), + }; + emit_ai_chat_event(&app, ev); + } Ok("ok".to_string()) } @@ -2545,6 +2600,8 @@ pub async fn ai_chat_stop( prompt_cache_hit_tokens: 0, prompt_cache_miss_tokens: 0, reasoning_tokens: 0, // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, + // 零 token 收敛信号,非截断(看板 953;枚举新字段,Rust 字面量强制显式赋值)。 + truncated_by_max_tokens: false, incomplete: None, incomplete_reason: None, conversation_id: Some(target.clone()), @@ -2666,6 +2723,8 @@ pub async fn ai_chat_stop( reasoning_tokens: 0, // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, + // 零 token 收敛信号,非截断(看板 953;枚举新字段,Rust 字面量强制显式赋值)。 + truncated_by_max_tokens: false, incomplete: None, incomplete_reason: None, conversation_id: conv_id.clone(), @@ -2864,6 +2923,8 @@ pub async fn ai_stop_loop( reasoning_tokens: 0, // 零 token 收敛信号,非估算(无真实 LLM 调用)。 is_estimated: false, + // 零 token 收敛信号,非截断(看板 953;枚举新字段,Rust 字面量强制显式赋值)。 + truncated_by_max_tokens: false, incomplete: None, incomplete_reason: None, conversation_id: Some(conversation_id), diff --git a/src-tauri/src/commands/ai/commands/conversation.rs b/src-tauri/src/commands/ai/commands/conversation.rs index 970cfbd..17e6378 100644 --- a/src-tauri/src/commands/ai/commands/conversation.rs +++ b/src-tauri/src/commands/ai/commands/conversation.rs @@ -79,6 +79,11 @@ pub fn record_to_message(rec: &AiMessageRecord) -> ChatMessage { .as_deref() .filter(|s| !s.is_empty()) .and_then(|s| serde_json::from_str(s).ok()); + let source = rec + .source + .as_deref() + .filter(|s| !s.is_empty()) + .and_then(|s| serde_json::from_str(s).ok()); let tool_calls = rec .tool_calls .as_deref() @@ -107,6 +112,7 @@ pub fn record_to_message(rec: &AiMessageRecord) -> ChatMessage { prompt_cache_miss_tokens: rec.prompt_cache_miss_tokens, reasoning_tokens: rec.reasoning_tokens, is_estimated: rec.is_estimated, + source, } } @@ -137,6 +143,10 @@ pub fn message_to_record( .as_ref() .map(|s| s.as_db_str().to_string()) .unwrap_or_else(|| "active".to_string()); + let source = msg + .source + .as_ref() + .map(|s| serde_json::to_string(s).unwrap_or_default()); AiMessageRecord { id, conversation_id: conversation_id.to_string(), @@ -157,6 +167,7 @@ pub fn message_to_record( prompt_cache_miss_tokens: msg.prompt_cache_miss_tokens, reasoning_tokens: msg.reasoning_tokens, is_estimated: msg.is_estimated, + source, } } @@ -364,6 +375,7 @@ pub(crate) async fn recover_parked_inner( None, // parts:纯文本 None, // mention_spans true, // pin_active:恢复动作即用户当前意图 + None, // #958:恢复路径无外部来源 ) .await } @@ -922,6 +934,8 @@ pub async fn ai_conversation_delete( let zero_usage = df_ai::provider::TokenUsage::default(); crate::commands::ai::agentic::emit_ai_completed_once( &app, &conversation_id, &zero_usage, false, + // 会话删除终态补发,非截断(看板 953)。 + false, Some(true), None, Some("deleted"), true, &[], false, ).await; @@ -1176,6 +1190,7 @@ mod tests { id: Some("msg_test_1".into()), role: MessageRole::User, content: "你好".into(), + source: None, parts: None, tool_call_id: None, tool_calls: None, diff --git a/src-tauri/src/commands/ai/conversation.rs b/src-tauri/src/commands/ai/conversation.rs index 281ddcc..01989f9 100644 --- a/src-tauri/src/commands/ai/conversation.rs +++ b/src-tauri/src/commands/ai/conversation.rs @@ -788,6 +788,7 @@ mod tests { id: None, role, content: content.to_string(), + source: None, parts: None, tool_call_id: None, tool_calls: None, diff --git a/src-tauri/src/commands/ai/outbox.rs b/src-tauri/src/commands/ai/outbox.rs index 508577d..d6faaed 100644 --- a/src-tauri/src/commands/ai/outbox.rs +++ b/src-tauri/src/commands/ai/outbox.rs @@ -10,8 +10,9 @@ //! - 单会话单轮一条:同会话同轮只派一条(串行保序),会话忙则本轮跳过(不丢,下轮再取) //! - 开关默认关:未配置 `df-outbox-enabled` 时循环只读配置空转,零行为 //! -//! ⚠️ P0 地基:调度器与派发链已就位,但 **未接线**(lib.rs setup 挂载 + agentic 收敛出口钩子), -//! 故本模块暂整体 allow(dead_code)。接线清单见交付说明,不在此文件内改 lib.rs。 +//! ⚠️ P0 地基 + P1 IPC 命令族均已就位,但 **未接线**(lib.rs setup 挂载 + generate_handler 注册 +//! + agentic 收敛出口钩子),故本模块暂整体 allow(dead_code)。接线清单见交付说明, +//! 不在此文件内改 lib.rs。 //! //! 配置面(Settings KV,通用 settings_get/settings_set IPC 可读写): //! - `df-outbox-enabled` bool("true"/"1" 为开),默认 **false**(地基默认关,可观测空转) @@ -23,10 +24,11 @@ use std::collections::HashSet; use std::sync::OnceLock; use std::time::Duration; -use tauri::{AppHandle, Emitter, Manager}; +use tauri::{AppHandle, Emitter, Manager, State}; use df_ai::provider::ContentPart; -use df_storage::crud::OutboxRepo; +use df_storage::crud::{OutboxRepo, STATUS_CANCELED, STATUS_FAILED, STATUS_PENDING, STATUS_SENDING, STATUS_SENT}; +use df_storage::models::AiOutboxRecord; use df_types::augmentation::MentionSpanDto; use crate::commands::ai::commands::chat::spawn_agentic_session; @@ -195,65 +197,88 @@ async fn run_tick(app: &AppHandle, cfg: &OutboxConfig) { continue; } } - let parts = parse_parts(item.parts_json.as_deref()); - let spans = parse_spans(item.spans_json.as_deref()); - let result = spawn_agentic_session( - app, - &state, - item.text.clone(), - None, // language 沿用默认 - item.skill.clone(), // /技能名 - item.model_override.clone(), - Some(conv.clone()), // 目标会话(定时/队列不误发到活跃会话) - parts, - spans, - false, // pin_active=false:不劫持用户活跃面板 - ) - .await; - match result { - Ok(_conv_id) => { - // Ok=受理成功即终态:轮次内失败(流断/工具错)由既有 loop 自愈,不回流 outbox。 - let _ = repo.mark_sent(&item.id).await; - dispatched.insert(conv.clone()); - emit(app, "ai-outbox-sent", &item.id, &conv, None, None); - emit(app, "ai-outbox-changed", &item.id, &conv, None, None); - tracing::info!("[outbox] 派发 {} → 会话 {conv}", item.id); - } - Err(err) => { - if is_permanent_spawn_error(&err) { + // 派发 + 结果落状态收口在 dispatch_claimed(调度循环与 send_now 共用一条派发链)。 + if dispatch_claimed(app, &state, &repo, cfg.max_attempts, item) + .await + .is_ok() + { + dispatched.insert(conv); + } + } +} + +/// 派发单条已 claim 的项:spawn 正典链 + 按受理结果落状态。调度循环(run_tick)与 +/// 立即派发(ai_outbox_send_now)共用,保证两条入口的失败分流/事件/通知语义一致。 +/// +/// Ok=受理成功(已 mark_sent);Err=受理失败的原始错误串(状态已按永久/瞬时落好)。 +async fn dispatch_claimed( + app: &AppHandle, + state: &AppState, + repo: &OutboxRepo, + max_attempts: i64, + item: AiOutboxRecord, +) -> Result { + let conv = item.conversation_id.clone(); + let parts = parse_parts(item.parts_json.as_deref()); + let spans = parse_spans(item.spans_json.as_deref()); + let result = spawn_agentic_session( + app, + state, + item.text.clone(), + None, // language 沿用默认 + item.skill.clone(), // /技能名 + item.model_override.clone(), + Some(conv.clone()), // 目标会话(定时/队列不误发到活跃会话) + parts, + spans, + false, // pin_active=false:不劫持用户活跃面板 + None, // #958:待发箱消息原桌面发起(来源已在原消息),续投不重标 + ) + .await; + match result { + Ok(conv_id) => { + // Ok=受理成功即终态:轮次内失败(流断/工具错)由既有 loop 自愈,不回流 outbox。 + let _ = repo.mark_sent(&item.id).await; + emit(app, "ai-outbox-sent", &item.id, &conv, None, None); + emit(app, "ai-outbox-changed", &item.id, &conv, None, None); + tracing::info!("[outbox] 派发 {} → 会话 {conv}", item.id); + Ok(conv_id) + } + Err(err) => { + if is_permanent_spawn_error(&err) { + let _ = repo.mark_failed(&item.id, &err).await; + emit(app, "ai-outbox-failed", &item.id, &conv, Some(&err), Some(true)); + notify_session_event( + app, + NotifyKind::Failed, + &conv, + "定时/待发消息发送失败", + &format!("受理失败(不可重试): {err}"), + ); + tracing::error!("[outbox] 受理永久失败 {}: {err}", item.id); + } else { + // 瞬时失败:回退 pending 保留下轮再试(网络超时/会话忙竞态天然重试)。 + let attempts = repo.revert_pending(&item.id, Some(&err)).await.unwrap_or(0); + if attempts >= max_attempts { let _ = repo.mark_failed(&item.id, &err).await; - emit(app, "ai-outbox-failed", &item.id, &conv, Some(&err), Some(true)); + emit(app, "ai-outbox-failed", &item.id, &conv, Some(&err), Some(false)); notify_session_event( app, NotifyKind::Failed, &conv, "定时/待发消息发送失败", - &format!("受理失败(不可重试): {err}"), + &format!("连续 {attempts} 次受理失败: {err}"), ); - tracing::error!("[outbox] 受理永久失败 {}: {err}", item.id); + tracing::error!("[outbox] 重试达上限({attempts})转失败 {}: {err}", item.id); } else { - // 瞬时失败:回退 pending 保留下轮再试(网络超时/会话忙竞态天然重试)。 - let attempts = repo.revert_pending(&item.id, Some(&err)).await.unwrap_or(0); - if attempts >= cfg.max_attempts { - let _ = repo.mark_failed(&item.id, &err).await; - emit(app, "ai-outbox-failed", &item.id, &conv, Some(&err), Some(false)); - notify_session_event( - app, - NotifyKind::Failed, - &conv, - "定时/待发消息发送失败", - &format!("连续 {attempts} 次受理失败: {err}"), - ); - tracing::error!("[outbox] 重试达上限({attempts})转失败 {}: {err}", item.id); - } else { - emit(app, "ai-outbox-changed", &item.id, &conv, None, None); - tracing::warn!( - "[outbox] 受理瞬时失败 {} (第 {attempts} 次),下轮重试: {err}", - item.id - ); - } + emit(app, "ai-outbox-changed", &item.id, &conv, None, None); + tracing::warn!( + "[outbox] 受理瞬时失败 {} (第 {attempts} 次),下轮重试: {err}", + item.id + ); } } + Err(err) } } } @@ -308,7 +333,283 @@ fn emit( } // ============================================================ -// 单测(纯函数:配置解析/永久失败判据) +// IPC 命令族(P1 应用层) +// ⚠️ 未注册:generate_handler 接线清单见交付说明(不在此文件内改 lib.rs)。 +// Repo 经 State 现构(与调度器 outbox_repo() 同源,最小改动不加 state 字段)。 +// ============================================================ + +/// IPC 返回的待发项视图。字段与 `AiOutboxRecord` 同形(snake_case 原样透传, +/// 前端 TS interface 按同名对齐),仅收窄为对外可见面(隐去 parts/spans 原始 JSON)。 +#[derive(Debug, Clone, serde::Serialize)] +pub struct AiOutboxItem { + pub id: String, + pub conversation_id: String, + pub kind: String, + pub text: String, + pub skill: Option, + pub model_override: Option, + pub due_at: Option, + pub status: String, + pub attempts: i64, + pub last_error: Option, + pub created_at: i64, + pub updated_at: i64, + pub sent_at: Option, +} + +impl From for AiOutboxItem { + fn from(r: AiOutboxRecord) -> Self { + Self { + id: r.id, + conversation_id: r.conversation_id, + kind: r.kind, + text: r.text, + skill: r.skill, + model_override: r.model_override, + due_at: r.due_at, + status: r.status, + attempts: r.attempts, + last_error: r.last_error, + created_at: r.created_at, + updated_at: r.updated_at, + sent_at: r.sent_at, + } + } +} + +// ── 参数校验(纯函数,单测直接覆盖;命令体只做校验→落库→事件三步) ── + +/// text 非空(enqueue / update_text 共用;空白串视为空)。 +fn validate_text(text: &str) -> Result<(), String> { + if text.trim().is_empty() { + return Err("消息内容不能为空".to_string()); + } + Ok(()) +} + +/// conversation_id 非空(设计 §3.1 目标会话 NOT NULL,composer 处确定,后端不猜 active)。 +fn validate_conv_id(conv: &str) -> Result<(), String> { + if conv.trim().is_empty() { + return Err("目标会话不能为空".to_string()); + } + Ok(()) +} + +/// enqueue 的 due_at 必须严格晚于当前时刻(定时项到点即发,过去/当下时刻无意义)。 +/// 注意:ai_outbox_reschedule **不做**此限制——改期到过去=「下轮立即派发」语义,属合法操作。 +fn validate_due_at_future(due_at: i64, now_ms: i64) -> Result<(), String> { + if due_at <= now_ms { + return Err(format!("定时时间必须晚于当前时刻(due_at={due_at}, now={now_ms})")); + } + Ok(()) +} + +/// kind 由 due_at 有无唯一推导(queue=尽快 / scheduled=定时),单一推导点防裸串漂移。 +fn kind_for_due(due_at: Option) -> &'static str { + if due_at.is_some() { + df_storage::crud::KIND_SCHEDULED + } else { + df_storage::crud::KIND_QUEUE + } +} + +/// send_now 状态守卫:pending 直接派、failed 先重入队再派;在途/终态拒绝并说明原因。 +fn ensure_sendable(status: &str) -> Result<(), String> { + match status { + STATUS_PENDING | STATUS_FAILED => Ok(()), + STATUS_SENDING => Err("该项正在发送中,请稍后".to_string()), + STATUS_SENT => Err("该项已发送成功,无需重复发送".to_string()), + STATUS_CANCELED => Err("该项已取消,无法发送".to_string()), + other => Err(format!("未知状态,拒绝派发: {other}")), + } +} + +/// cancel 状态守卫:sent(已发出去的消息)不可「取消」,canceled 幂等拒绝;其余可取消。 +fn ensure_cancellable(status: &str) -> Result<(), String> { + match status { + STATUS_SENT => Err("该项已发送成功,无法取消".to_string()), + STATUS_CANCELED => Err("该项已处于取消状态".to_string()), + _ => Ok(()), + } +} + +/// 入队一条待发/定时消息。`due_at`=Some(epoch ms,必须未来)→ 定时项;None → 尽快队列项。 +/// 返回新项 id。成功即唤醒调度器(尽快项可被立刻取走)并 emit ai-outbox-changed。 +#[tauri::command] +pub async fn ai_outbox_enqueue( + app: AppHandle, + state: State<'_, AppState>, + text: String, + due_at: Option, + skill: Option, + model_override: Option, + conversation_id: String, +) -> Result { + validate_text(&text)?; + validate_conv_id(&conversation_id)?; + let now = df_types::now_millis(); + if let Some(due) = due_at { + validate_due_at_future(due, now)?; + } + let id = df_types::types::new_id(); + let record = AiOutboxRecord { + id: id.clone(), + conversation_id: conversation_id.clone(), + kind: kind_for_due(due_at).to_string(), + text, + skill, + model_override, + // P1 定时入口纯文本;图片项待后续(parts 落库链已就绪,enqueue 参数面再扩)。 + parts_json: None, + spans_json: None, + due_at, + status: STATUS_PENDING.to_string(), + attempts: 0, + last_error: None, + // created_at/updated_at 由 Repo.insert 以落库时刻覆盖,此处仅占位。 + created_at: now, + updated_at: now, + sent_at: None, + }; + OutboxRepo::new(&state.db) + .insert(record) + .await + .map_err(|e| format!("待发项写入失败: {e}"))?; + // 尽快项/到期项即时唤醒调度器(周期 tick 仅兜底);事件供各窗口视图刷新。 + notify().notify_one(); + emit(&app, "ai-outbox-changed", &id, &conversation_id, None, None); + Ok(id) +} + +/// 列出待发项。`conv_id`=Some → 该会话;None → 跨会话全量(定时列表抽屉数据源)。 +/// 恒含终态(前端显 pending|sent|failed|canceled 状态徽标)。 +#[tauri::command] +pub async fn ai_outbox_list( + state: State<'_, AppState>, + conv_id: Option, +) -> Result, String> { + let repo = OutboxRepo::new(&state.db); + let records = match conv_id { + Some(conv) if !conv.trim().is_empty() => repo.list_by_conv(&conv, true).await, + _ => repo.list_all(true).await, + }; + let records = records.map_err(|e| format!("待发项查询失败: {e}"))?; + Ok(records.into_iter().map(AiOutboxItem::from).collect()) +} + +/// 取消待发项。已发送/已取消的项拒绝(幂等不假报成功)。 +#[tauri::command] +pub async fn ai_outbox_cancel( + app: AppHandle, + state: State<'_, AppState>, + id: String, +) -> Result { + let repo = OutboxRepo::new(&state.db); + let item = repo + .get_by_id(&id) + .await + .map_err(|e| format!("待发项查询失败: {e}"))? + .ok_or_else(|| format!("待发项不存在: {id}"))?; + ensure_cancellable(&item.status)?; + let ok = repo.cancel(&id).await.map_err(|e| format!("取消失败: {e}"))?; + if ok { + emit(&app, "ai-outbox-changed", &id, &item.conversation_id, None, None); + } + Ok(ok) +} + +/// 改期。允许过去时刻(=下轮立即派发语义,与 enqueue 的未来约束不同,见 validate_due_at_future 注)。 +#[tauri::command] +pub async fn ai_outbox_reschedule( + app: AppHandle, + state: State<'_, AppState>, + id: String, + due_at: i64, +) -> Result { + let repo = OutboxRepo::new(&state.db); + let item = repo + .get_by_id(&id) + .await + .map_err(|e| format!("待发项查询失败: {e}"))? + .ok_or_else(|| format!("待发项不存在: {id}"))?; + let ok = repo + .reschedule(&id, Some(due_at)) + .await + .map_err(|e| format!("改期失败: {e}"))?; + if ok { + // 改期到已到点的新时刻可被立即取走,唤醒调度器。 + notify().notify_one(); + emit(&app, "ai-outbox-changed", &id, &item.conversation_id, None, None); + } + Ok(ok) +} + +/// 编辑待发文本。终态(sent/canceled)项拒绝——历史记录不可改写。 +#[tauri::command] +pub async fn ai_outbox_update_text( + app: AppHandle, + state: State<'_, AppState>, + id: String, + text: String, +) -> Result { + validate_text(&text)?; + let repo = OutboxRepo::new(&state.db); + let item = repo + .get_by_id(&id) + .await + .map_err(|e| format!("待发项查询失败: {e}"))? + .ok_or_else(|| format!("待发项不存在: {id}"))?; + if item.status == STATUS_SENT || item.status == STATUS_CANCELED { + return Err(format!("该项已终态({}),不可编辑", item.status)); + } + let ok = repo + .update_text(&id, &text) + .await + .map_err(|e| format!("编辑失败: {e}"))?; + if ok { + emit(&app, "ai-outbox-changed", &id, &item.conversation_id, None, None); + } + Ok(ok) +} + +/// 立即派发一次:状态守卫 →(failed 先重入队)→ CAS claim → 走与调度器同一条派发链 +/// (dispatch_claimed,失败分流/事件/通知语义完全一致)。 +/// 显式用户动作,不受 `df-outbox-enabled` 开关约束(开关只管后台自动调度)。 +#[tauri::command] +pub async fn ai_outbox_send_now( + app: AppHandle, + state: State<'_, AppState>, + id: String, +) -> Result<(), String> { + let repo = OutboxRepo::new(&state.db); + let item = repo + .get_by_id(&id) + .await + .map_err(|e| format!("待发项查询失败: {e}"))? + .ok_or_else(|| format!("待发项不存在: {id}"))?; + ensure_sendable(&item.status)?; + if item.status == STATUS_FAILED { + let requeued = repo.requeue(&id).await.map_err(|e| format!("重入队失败: {e}"))?; + if !requeued { + return Err("重入队失败(状态已变化),请刷新后重试".to_string()); + } + } + if !repo.claim(&id).await.map_err(|e| format!("占取失败: {e}"))? { + return Err("该项已被调度占取,请稍后刷新".to_string()); + } + // 占取后重取最新记录(sending 态),交给共用派发链落状态。 + let fresh = repo + .get_by_id(&id) + .await + .map_err(|e| format!("待发项查询失败: {e}"))? + .ok_or_else(|| format!("待发项不存在: {id}"))?; + let max_attempts = read_config(&app).await.max_attempts; + dispatch_claimed(&app, &state, &repo, max_attempts, fresh).await?; + Ok(()) +} + +// ============================================================ +// 单测(纯函数:配置解析/永久失败判据/命令参数校验与错误映射) // ============================================================ #[cfg(test)] @@ -368,4 +669,65 @@ mod tests { fn status_constant_single_source() { assert_eq!(df_storage::crud::STATUS_PENDING, "pending"); } + + // ── P1 IPC 命令族:参数校验与错误映射(命令体的纯函数面) ── + + /// enqueue/update_text 的 text 校验:空白串拒绝且错误信息明确,非空放行。 + #[test] + fn validate_text_rejects_blank_with_message() { + assert!(validate_text("").is_err()); + assert!(validate_text(" \n\t ").is_err()); + assert!(validate_text("跑一下构建").is_ok()); + let err = validate_text(" ").unwrap_err(); + assert!(err.contains("不能为空"), "错误信息应可直读: {err}"); + } + + /// enqueue 的 conversation_id 校验:空/空白拒绝(目标会话 NOT NULL,后端不猜 active)。 + #[test] + fn validate_conv_id_rejects_blank() { + assert!(validate_conv_id("").is_err()); + assert!(validate_conv_id(" ").is_err()); + assert!(validate_conv_id("conv-abc").is_ok()); + } + + /// enqueue 的 due_at 校验:过去/当下时刻拒绝,严格未来放行; + /// reschedule 特意不走此校验(过去时刻=立即派发语义),故仅 enqueue 面约束。 + #[test] + fn validate_due_at_future_rejects_past_and_present() { + let now = 1_000_000; + assert!(validate_due_at_future(now - 1, now).is_err(), "过去时刻拒绝"); + assert!(validate_due_at_future(now, now).is_err(), "当下时刻(非严格未来)拒绝"); + assert!(validate_due_at_future(now + 1, now).is_ok()); + let err = validate_due_at_future(42, 43).unwrap_err(); + assert!(err.contains("必须晚于当前时刻"), "错误信息应可直读: {err}"); + } + + /// kind 单一推导点:due_at 有无唯一决定 scheduled/queue(防两处裸串漂移)。 + #[test] + fn kind_for_due_maps_by_presence() { + assert_eq!(kind_for_due(Some(9_999_999_999_999)), df_storage::crud::KIND_SCHEDULED); + assert_eq!(kind_for_due(None), df_storage::crud::KIND_QUEUE); + } + + /// send_now 状态守卫错误映射:pending/failed 可派;在途与各终态逐一拒绝且信息可直读。 + #[test] + fn sendable_guard_maps_status_errors() { + assert!(ensure_sendable(STATUS_PENDING).is_ok()); + assert!(ensure_sendable(STATUS_FAILED).is_ok(), "failed 经 requeue 后可手动重派"); + assert!(ensure_sendable(STATUS_SENDING).is_err()); + assert!(ensure_sendable(STATUS_SENT).is_err()); + assert!(ensure_sendable(STATUS_CANCELED).is_err()); + assert!(ensure_sendable("bogus").is_err(), "未知状态保守拒绝"); + assert!(ensure_sendable(STATUS_SENT).unwrap_err().contains("已发送成功")); + } + + /// cancel 状态守卫:sent 不可取消(消息已出)、canceled 幂等拒绝;pending/failed/sending 可取消。 + #[test] + fn cancellable_guard_blocks_terminal_only() { + assert!(ensure_cancellable(STATUS_PENDING).is_ok()); + assert!(ensure_cancellable(STATUS_SENDING).is_ok(), "在途也允许取消意图(取消落 canceled)"); + assert!(ensure_cancellable(STATUS_FAILED).is_ok(), "失败项可放弃取消"); + assert!(ensure_cancellable(STATUS_SENT).is_err()); + assert!(ensure_cancellable(STATUS_CANCELED).is_err()); + } } diff --git a/src-tauri/src/commands/ai/title.rs b/src-tauri/src/commands/ai/title.rs index 4bd6de9..e27c7a2 100644 --- a/src-tauri/src/commands/ai/title.rs +++ b/src-tauri/src/commands/ai/title.rs @@ -78,6 +78,7 @@ pub(crate) async fn ensure_conversation_title( id: None, // 标题摘要派生消息:不落库不溯源,无需分配 ID(对齐老消息 None 语义) role: m.role.clone(), content: m.content.clone(), + source: None, parts: None, tool_call_id: None, tool_calls: None, diff --git a/src-tauri/src/commands/ai/unattended.rs b/src-tauri/src/commands/ai/unattended.rs index 0332f04..c59a51c 100644 --- a/src-tauri/src/commands/ai/unattended.rs +++ b/src-tauri/src/commands/ai/unattended.rs @@ -1300,6 +1300,7 @@ async fn recover_one_parked(app: &AppHandle) -> Result { None, None, false, + None, // #958:引擎路径无外部来源 ) .await .map(|_| conv_id) @@ -1363,6 +1364,7 @@ async fn claim_and_spawn(app: &AppHandle, cfg: &EngineConfig) -> Result<(), Stri None, None, false, // pin_active=false 不劫持活跃面板 + None, // #958:引擎路径无外部来源 ) .await { diff --git a/src-tauri/src/commands/ai/wake.rs b/src-tauri/src/commands/ai/wake.rs index a217f22..70028fd 100644 --- a/src-tauri/src/commands/ai/wake.rs +++ b/src-tauri/src/commands/ai/wake.rs @@ -231,6 +231,7 @@ async fn fire(app: &AppHandle, entry: &WakeEntry) -> bool { None, None, false, // pin_active=false:唤醒不劫持用户活跃面板 + None, // #958:唤醒路径无外部来源 ) .await { diff --git a/src-tauri/src/feishu_channel.rs b/src-tauri/src/feishu_channel.rs index fe138fc..4f24c41 100644 --- a/src-tauri/src/feishu_channel.rs +++ b/src-tauri/src/feishu_channel.rs @@ -22,6 +22,7 @@ use serde::Deserialize; use df_storage::crud::SettingsRepo; use df_types::types::new_id; +use df_ai::provider::ContentPart; /// app_settings 键:飞书渠道配置(与设置面板 FeishuPanel 同键契约) const KV_FEISHU_CONFIG: &str = "feishu-config"; @@ -371,6 +372,103 @@ fn parse_command(text: &str) -> Option<(String, String)> { Some((cmd, tail)) } +/// 图片消息喂模型的任务语境(#958):飞书 image 为独立消息类型无法带文字, +/// 用引导词让 vision 模型知道「这是用户发来的图,按图作答」 +const IMAGE_INBOUND_PROMPT: &str = "(用户通过飞书发来一张图片,请查看图片并按图片内容处理)"; + +/// 图片入站大小上限(#958):base64 转换膨胀 4/3,vision 端点普遍接受 ≤20MB 内嵌图, +/// 10MB 原图留足余量;超限引导用户压缩,防超大图烧 token 与转码内存 +const MAX_IMAGE_BYTES: usize = 10 * 1024 * 1024; + +/// 下载飞书图片消息资源并落盘,返回本地引用 ContentPart(#958 引用化存储)。 +/// 存储面只存 `file://` 引用(base64 不进 SQLite,防大图随历史全量重放膨胀), +/// 发送请求时由 df-ai::image_ref 单点按需读盘转 base64(模型使用按需转换)。 +/// 走 channel_messaging 便捷方法(与 reply/send 同风格),租户 token 由 client 自管。 +/// media_type 由字节魔数判定(下载响应无可靠 MIME;飞书图床为 png/jpeg/webp/gif), +/// 未知格式拒收——错误 media_type 会被 vision 端点 400,宁早拒不晚炸。 +async fn download_feishu_image( + client: &LarkClient, + store: &crate::media_store::MediaStore, + message_id: &str, + file_key: &str, +) -> Result { + // 下载带防线(对齐 u-claw fetchResource 经验):3 次指数退避(1s/4s)+单次 120s 超时 + // ——飞书资源接口偶发抖动,无重试时用户收到一次性失败;超时防接口挂起拖死受理链。 + // 空响应与错误同待遇进重试(可能为服务端瞬时异常),耗尽后报最后一次原因 + let resp = { + let mut last_err = String::from("下载失败: 未知错误"); + let mut out = None; + for attempt in 0u32..3 { + if attempt > 0 { + tokio::time::sleep(std::time::Duration::from_secs(1 << (attempt - 1))).await; + } + let dl = tokio::time::timeout( + std::time::Duration::from_secs(120), + client.channel_messaging().download_message_resource( + message_id, + file_key, + "image", + &RequestOption::default(), + ), + ) + .await; + match dl { + Ok(Ok(r)) if !r.data.is_empty() => { + out = Some(r); + break; + } + Ok(Ok(_)) => last_err = "图片内容为空".into(), + Ok(Err(e)) => last_err = format!("下载失败: {e}"), + Err(_) => last_err = "下载超时(120s)".into(), + } + } + match out { + Some(r) => r, + None => return Err(last_err), + } + }; + if resp.data.len() > MAX_IMAGE_BYTES { + return Err(format!( + "图片超过 10MB 上限(约 {}MB),请压缩后重发", + resp.data.len() / 1024 / 1024 + )); + } + let media_type = sniff_image_mime(&resp.data) + .ok_or_else(|| "不支持的图片格式(仅支持 png/jpeg/webp/gif)".to_string())?; + // 存储决策收口 media_store:hash 去重落盘,OSS 配置在则双写并引用 https(跨设备可解析), + // 失败降级本机 file://(不阻塞)。本函数只管飞书协议(下载+防线),不管存储。 + // 飞书 image 消息无原始文件名(None=纯 hash 名) + let url = store.save(&resp.data, media_type, None).await?; + Ok(ContentPart::Image { + url: Some(url), + base64: None, + media_type: Some(media_type.into()), + alt: None, + }) +} + +/// 读取媒体 OSS 配置(KV `media-oss-config`;未配置/解析失败 → None=纯本地模式) +async fn read_media_oss_config( + db: &df_storage::db::Database, +) -> Option { + crate::media_store::read_oss_config(db).await +} + +/// 图片格式魔数判定(#958):PNG/JPEG/GIF/WEBP 四态,与飞书图床常见格式对齐 +fn sniff_image_mime(b: &[u8]) -> Option<&'static str> { + if b.starts_with(&[0x89, b'P', b'N', b'G']) { + Some("image/png") + } else if b.starts_with(&[0xFF, 0xD8, 0xFF]) { + Some("image/jpeg") + } else if b.starts_with(b"GIF8") { + Some("image/gif") + } else if b.len() > 12 && b.starts_with(b"RIFF") && &b[8..12] == b"WEBP" { + Some("image/webp") + } else { + None + } +} + /// 入站文本:群聊需 @bot;白名单校验;/new 重置;其余走 ai-chat 会话链路 async fn handle_message( ctx: Arc, @@ -378,23 +476,47 @@ async fn handle_message( ) { // D3:任何入站消息都是事件管道存活证据,先于一切过滤更新活性时间戳 ctx.last_alive.store(now_secs(), Ordering::SeqCst); - if msg.message_type != "text" { + // 类型分流(#958 图片入站):text 原样;image 走下载装配(vision ContentPart); + // 其余类型维持 #25 提示卡。image 消息 SDK 归一化已把 image_key 放入 + // msg.resources(ChannelResource),无需自行解析 content JSON。 + let image_key = if msg.message_type == "image" { + match msg.resources.iter().find(|r| r.resource_type == "image") { + Some(r) => Some(r.file_key.clone()), + None => { + let _ = reply_markdown_card( + &ctx.client, + &msg.message_id, + "⚠️ 图片消息缺少 image_key,无法处理。", + ) + .await; + return; + } + } + } else if msg.message_type != "text" { // #25 非文本消息静默丢修复:回提示卡(D7 媒体能力前的兜底),不再零反馈。 - // 补替代路径引导:用户发图/文件往往是想让 AI 处理内容,直接告知改用文字描述 + // 补替代路径引导:用户发文件等往往是想让 AI 处理内容,直接告知改用文字描述 let _ = reply_markdown_card( &ctx.client, &msg.message_id, - "📎 暂仅支持文本消息,图片/文件等媒体能力规划中。\n\n如需处理图片/文件内容,请用文字描述你的需求(如「把这段文字翻译成英文:…」)。", + "📎 暂支持文本与图片消息,文件/语音等其他类型规划中。\n\n如需处理文件内容,请用文字描述你的需求(如「把这段文字翻译成英文:…」)。", ) .await; return; - } + } else { + None + }; if msg.chat_type == "group" && !msg.mentioned_bot { return; } - let text = msg.text.clone().unwrap_or_default().trim().to_string(); - // 飞书 @ 占位清洗(如 @_user_1):@ 语义已由 mentioned_bot 承载,占位串只污染文本 - let text = strip_mention_placeholders(&text); + // 图片消息不带文字(飞书 image 为独立消息类型),喂固定引导词交代任务语境; + // msg.text 的 markdown 图链([图片](image_key))对模型无意义,不喂 + let text = if image_key.is_some() { + IMAGE_INBOUND_PROMPT.to_string() + } else { + let text = msg.text.clone().unwrap_or_default().trim().to_string(); + // 飞书 @ 占位清洗(如 @_user_1):@ 语义已由 mentioned_bot 承载,占位串只污染文本 + strip_mention_placeholders(&text) + }; if text.is_empty() { return; } @@ -427,6 +549,30 @@ async fn handle_message( return; } + // 图片入站(#958):下载消息资源→落盘→引用片。置于白名单后(未授权不下载, + // 不替白名单外用户烧流量)、占位发票前(失败早退,无占位卡需原地更新) + let parts = match &image_key { + Some(key) => { + let state = ctx.app.state::(); + let oss_cfg = read_media_oss_config(&state.db).await; + let store = crate::media_store::MediaStore::new(state.data_dir.clone(), oss_cfg); + match download_feishu_image(&ctx.client, &store, &message_id, key).await { + Ok(part) => Some(vec![part]), + Err(e) => { + feishu_log(&format!("图片下载失败 chat={chat_id}: {e}")); + let _ = reply_markdown_card( + &ctx.client, + &message_id, + &format!("⚠️ 图片下载失败:{e}\n\n请重发,或直接用文字描述你的需求。"), + ) + .await; + return; + } + } + } + None => None, + }; + // 命令通路(#30 容错):全角/大小写/附言归一后识别,仅命令词走控制面,任务原文不动。 // 顺序约束:命令与审批决议须先于注入旁路——conv 活跃时这两类文本否则会被当补充 // 指令注入任务;未知命令词落回常规任务路径,零行为变更 @@ -578,7 +724,9 @@ async fn handle_message( let state = ctx.app.state::(); let conv_id_now = ctx.chat_map.lock().await.get(&chat_id).cloned(); if let Some(cid) = conv_id_now { - if state.conv_states.is_active(&cid) { + // 图片消息不注入:inject_runtime_message 只收文本,注入则图被丢、模型只见 + // 引导词空跑;落常规路径由受理守卫拦,语义诚实(任务占用中,稍后重发) + if image_key.is_none() && state.conv_states.is_active(&cid) { match crate::commands::ai::commands::chat::inject_runtime_message(&state, &cid, &text).await { Ok(true) => { let _ = reply_markdown_card(&ctx.client, &message_id, "✅ 补充指令已注入运行中任务,下一轮生效").await; @@ -654,9 +802,16 @@ async fn handle_message( None, None, Some(conv_id.clone()), - None, + parts, // #958 图片入站:图片消息带 vision 片,文本消息 None;vision 能力判定/降级在 spawn 内 None, false, // 不劫持用户活跃面板(编排节点同款语义) + // #958 溯源:飞书原始消息定位随 user 消息落库,可跳原消息/区分多端来源 + Some(df_ai::provider::MessageSource { + channel: "feishu".into(), + message_id: message_id.clone(), + chat_id: Some(chat_id.clone()), + sender: if open_id.is_empty() { None } else { Some(open_id.clone()) }, + }), ) .await { @@ -717,10 +872,12 @@ async fn handle_message( let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>(); cancel_watchers().lock().await.insert(conv_id.clone(), cancel_tx); let conv_watched = conv_id.clone(); + let watched_chat = chat_id.clone(); tauri::async_runtime::spawn(async move { run_completion_watcher( ctx, conv_id, + watched_chat, ack, inflight_guard, cancel_rx, @@ -745,6 +902,145 @@ struct CardFrame { terminal: bool, } +// ── 出站文件回传(#958,u-claw replyLocalFiles 同模式) ── +// 约定:AI 报告中写 `[交付] <绝对路径>` 行,渠道自动把文件上传飞书回发—— +// 图片走图片消息、其余走文件消息(飞书原生附件体验,免用户去桌面端翻)。 +// 正文中提及路径但无标记行=叙述,不发送(防止口述路径误上传)。 + +/// 提取报告中全部 `[交付] <路径>` 行的路径(纯函数)。容错:全半角冒号后空格可选, +/// 路径含空格整行取(行内唯一路径假设,与 AI 约定一致) +fn extract_delivery_paths(text: &str) -> Vec { + text.lines() + .filter_map(|line| { + let line = line.trim(); + let rest = line.strip_prefix("[交付]")?; + let path = rest.trim().trim_start_matches(':').trim_start_matches('：').trim(); + (!path.is_empty()).then(|| path.to_string()) + }) + .collect() +} + +/// 敏感文件拦截(防御纵深,对齐 u-claw replyBlocked):名字命中凭据/密钥特征即拒, +/// AI 被诱导交付敏感文件时兜底。返回 Some(原因)=拦截 +fn delivery_blocked(path: &str) -> Option<&'static str> { + let lower = path.to_lowercase(); + const SENSITIVE: [&str; 8] = [ + ".env", "id_rsa", "credential", "password", "secret", "token", "api_key", "apikey", + ]; + SENSITIVE + .iter() + .find(|k| lower.contains(**k)) + .map(|_| "敏感文件(凭据/密钥特征)") +} + +/// 扩展名 → 飞书 file_type(上传接口必填;对齐飞书开放平台枚举) +fn feishu_file_type(ext: &str) -> &'static str { + match ext { + "pdf" => "pdf", + "xls" | "xlsx" | "csv" => "xls", + "doc" | "docx" | "rtf" => "doc", + "ppt" | "pptx" => "ppt", + "mp4" | "mov" | "avi" | "mkv" => "mp4", + "mp3" | "wav" | "m4a" | "aac" | "flac" => "mp3", + _ => "stream", + } +} + +/// 上传回传全部交付文件。单文件失败只提示不中断(部分成功优于整体失败); +/// 文件不存在静默跳过并记日志(过滤残余误匹配,"写了路径文件不在"经日志可查) +async fn reply_deliveries(client: &LarkClient, chat_id: &str, text: &str) { + for path in extract_delivery_paths(text) { + if let Some(why) = delivery_blocked(&path) { + feishu_log(&format!("回传跳过({why}): {path}")); + send_text(client, chat_id, &format!("⚠️ 已拦截交付请求:{path}({why})")).await; + continue; + } + let p = std::path::Path::new(&path); + match tokio::fs::metadata(p).await { + Ok(m) if m.is_file() => {} + _ => { + feishu_log(&format!("回传跳过(文件不存在或为目录): {path}")); + continue; + } + } + let bytes = match tokio::fs::read(p).await { + Ok(b) => b, + Err(e) => { + feishu_log(&format!("回传读取失败({path}): {e}")); + continue; + } + }; + let opt = &RequestOption::default(); + let ext = p + .extension() + .map(|e| e.to_string_lossy().to_lowercase()) + .unwrap_or_default(); + let is_image = matches!(ext.as_str(), "png" | "jpg" | "jpeg" | "gif" | "webp"); + let sent: Result<(), String> = if is_image { + // 图片:上传 image → image 消息直发(直发不做 reply,占位卡已是回复链, + // 附件直发即可,避免嵌套回复噪声) + match client.im().image.create("message", bytes, opt).await { + Ok(resp) => match resp.data.as_ref().and_then(|d| d.image_key.clone()) { + Some(key) => { + let input = SendInput { + msg_type: Some("image".into()), + image_key: Some(key), + chat_id: Some(chat_id.to_string()), + ..Default::default() + }; + client + .channel_messaging() + .send(&input, opt) + .await + .map(|_| ()) + .map_err(|e| format!("发送失败: {e}")) + } + None => Err("响应缺 image_key".into()), + }, + Err(e) => Err(format!("图片上传失败: {e}")), + } + } else { + // 文件:上传 file → file 消息直发(飞书好友互发文件同款体验) + let name = p + .file_name() + .map(|n| n.to_string_lossy().to_string()) + .unwrap_or_else(|| "file".into()); + match client + .im() + .file + .create(feishu_file_type(&ext), &name, None, bytes, opt) + .await + { + Ok(resp) => match resp.data.as_ref().and_then(|d| d.file_key.clone()) { + Some(key) => { + let input = SendInput { + msg_type: Some("file".into()), + file_key: Some(key), + chat_id: Some(chat_id.to_string()), + ..Default::default() + }; + client + .channel_messaging() + .send(&input, opt) + .await + .map(|_| ()) + .map_err(|e| format!("发送失败: {e}")) + } + None => Err("响应缺 file_key".into()), + }, + Err(e) => Err(format!("文件上传失败: {e}")), + } + }; + if let Err(e) = sent { + feishu_log(&format!("回传失败({path}): {e}")); + let base = path.rsplit(['/', '\\']).next().unwrap_or(path.as_str()); + send_text(client, chat_id, &format!("⚠️ 文件回传失败:{base}({e})")).await; + } else { + feishu_log(&format!("回传完成: {path}")); + } + } +} + /// 完成事件订阅任务(#14 两段式):只做裁决——消费 ai_event_bus 的 AiAgentFinished(本 conv) /// 决定终版文本,经帧通道交给预览任务原地落卡(占位卡唯一写者是预览任务,受理卡/终版/阶段 /// 提示全部经帧通道按序落卡)。引擎层 deliver_completion_report_if_waited 是广播单一发出点。 @@ -755,6 +1051,7 @@ struct CardFrame { async fn run_completion_watcher( ctx: Arc, conv_id: String, + chat_id: String, ack: String, // #20:计数生命周期跟任务走,guard 持有至 watcher 退出 _inflight: InflightGuard, @@ -855,6 +1152,10 @@ async fn run_completion_watcher( } }; feishu_log(&format!("watcher 终版 conv={conv_id} len={}", frame.text.len())); + // #958 出站回传(u-claw replyLocalFiles 同模式):报告内 [交付] 标记行的本地文件 + // 自动上传飞书(图片→图片消息/其余→文件消息),用户在飞书直接收原生附件。 + // 终版文本中的标记行本身保留(AI 写路径也是叙述的一部分,卡内可见) + reply_deliveries(&ctx.client, &chat_id, &frame.text).await; let _ = frame_tx.send(frame).await; let _ = stream_task.await; // 等预览任务落卡终版并退出,再释放 inflight return; @@ -1958,10 +2259,57 @@ async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap"), None); + assert_eq!(sniff_image_mime(&[]), None); + // RIFF 但非 WEBP(avi/wav 等)不误判 + assert_eq!(sniff_image_mime(b"RIFF\x24\x08\x00\x00AVI "), None); + } + #[test] fn command_case_and_tail_tolerance() { // 30:/new 大小写与附言容错,附言不烧 LLM diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 8dbff6b..9a9322a 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2,6 +2,8 @@ mod commands; mod state; +// 媒体存储域(#958 引用化):hash 去重落盘/OSS 双写/引用形态,飞书与桌面粘贴图共用 +mod media_store; // Phase3: 桌面内嵌 HTTP MCP server + 托盘常驻 + 单实例 mod mcp; // 本地工具代理统一网关:内嵌 axum 监听 127.0.0.1:8899,路由到 mysql/ssh/mongo/redis 4 代理端口 @@ -72,6 +74,9 @@ pub fn run() { crate::commands::ai::tools::spawn_agent::init_app_handle(app.handle().clone()); // D2a 工具结果外置根目录注入(app_data 下 tool-results/;未注入时直通降级) commands::ai::tool_result_offload::init_base(data_dir.clone()); + // P1 上下文侧工具结果外置(同根目录,tool-results/ 内以 `ctx-` 前缀区分)。 + // 未注入时 extract_key_info 退化为纯摘要 = 改造前行为(零回归)。 + df_ai::tool_offload::init_base(data_dir.clone()); // 飞书直连渠道:配置启用时起长连接(读 feishu-config;未配置/disabled 静默跳过) feishu_channel::start_if_enabled(app.handle().clone()); diff --git a/src-tauri/src/media_store.rs b/src-tauri/src/media_store.rs new file mode 100644 index 0000000..9e18172 --- /dev/null +++ b/src-tauri/src/media_store.rs @@ -0,0 +1,309 @@ +//! 媒体存储域(#958 图片入站引用化 + 跨设备文件服务) +//! +//! 单一归属:媒体字节 → 存储引用(`ContentPart::Image.url`)的全部决策收在本模块—— +//! hash 去重落盘(同内容同文件,转发/重发不重复存储)、可选 OSS 双写(跨设备可解析)、 +//! 引用形态选择(oss `https://` / 本机 `file://`)。飞书协议(下载/上传回传)留 feishu_channel, +//! 本模块零协议依赖。 +//! +//! 引用设计:不自造 scheme——`https://`(OSS)是 OpenAI 端点原生直传格式,anthropic 侧 +//! 已有 url→base64 预拉;`file://` 由 df-ai::image_ref 发送前展开为 base64。展开按 url +//! 形态自然分流,无自定义协议。 +//! +//! OSS 失败语义:上传失败不阻塞——本机快照已落盘,引用降级 `file://`(本机可解析, +//! 跨设备解析降级为「图片文件缺失」文本,诚实不炸请求)。 + +use df_ai::provider::ContentPart; +use serde::Deserialize; + +/// OSS 配置(app_settings KV `media-oss-config` JSON;缺省=纯本地模式)。 +/// public_base_url 为空时取 `https://{bucket}.{endpoint}`(bucket 私有时需自备 CDN 域) +#[derive(Debug, Clone, Deserialize)] +pub struct OssConfig { + pub bucket: String, + /// 如 oss-cn-hangzhou.aliyuncs.com + pub endpoint: String, + pub access_key_id: String, + pub access_key_secret: String, + #[serde(default)] + pub public_base_url: String, +} + +/// app_settings 键:媒体 OSS 配置(KV JSON,设置面板后续批;当前可手配) +pub const KV_MEDIA_OSS_CONFIG: &str = "media-oss-config"; + +/// 读 OSS 配置(KV;未配置/解析失败 → None=纯本地模式)。媒体域唯一读取口, +/// 飞书入站与 spawn offload 共用 +pub async fn read_oss_config(db: &df_storage::db::Database) -> Option { + let repo = df_storage::crud::SettingsRepo::new(db); + let raw = repo.get(KV_MEDIA_OSS_CONFIG).await.ok().flatten()?; + serde_json::from_str(&raw).ok() +} + +/// base64 图片片 → 落盘引用片(spawn 统一 offload,#958 引用化): +/// 桌面粘贴图/前端上传在此收口,消息只存引用,SQLite 不进大 base64 字符串 +/// (历史重放不再全量携带)。落盘失败保原 base64 片(不丢图,仅失去引用化收益); +/// 引用片/文本片原样透传。media_type 缺失的片无法规范落盘,原样保留 +pub async fn offload_image_parts(parts: Vec, store: &MediaStore) -> Vec { + use base64::Engine as _; + let mut out = Vec::with_capacity(parts.len()); + for p in parts { + out.push(match p { + ContentPart::Image { url, base64: Some(b64), media_type, alt } => { + let decoded = base64::engine::general_purpose::STANDARD + .decode(b64.as_bytes()) + .ok() + .filter(|b| !b.is_empty()); + match (decoded, media_type.clone()) { + (Some(bytes), Some(mt)) => match store.save(&bytes, &mt, None).await { + Ok(url) => ContentPart::Image { + url: Some(url), + base64: None, + media_type, + alt, + }, + Err(e) => { + tracing::warn!(err = %e, "图片落盘失败,保留 base64 片"); + ContentPart::Image { url, base64: Some(b64), media_type, alt } + } + }, + // 解码失败/空图/无 media_type:原样透传,provider 层按既有语义处置 + _ => ContentPart::Image { url, base64: Some(b64), media_type, alt }, + } + } + other => other, + }); + } + out +} + +/// 媒体存储。`oss` 为 None 时纯本地模式(引用全部 file://) +pub struct MediaStore { + data_dir: std::path::PathBuf, + oss: Option, +} + +impl MediaStore { + pub fn new(data_dir: std::path::PathBuf, oss: Option) -> Self { + Self { data_dir, oss } + } + + /// 落盘并返回引用 url。本地快照必落(hash 去重,同内容幂等);OSS 配置在则上传, + /// 成功引用 `https://`(跨设备可解析),失败降级 `file://`(不阻塞,本机零损)。 + /// + /// `original_name`:原始文件名(文件消息带,图片消息通常 None)——缀在 hash 后保 + /// 目录可读,清洗路径穿越字符防越界;None/空则纯 hash 名 + pub async fn save( + &self, + bytes: &[u8], + media_type: &str, + original_name: Option<&str>, + ) -> Result { + let key = content_key(bytes, media_type, original_name); + let dir = self.data_dir.join("media"); + tokio::fs::create_dir_all(&dir) + .await + .map_err(|e| format!("创建媒体目录失败: {e}"))?; + let path = dir.join(&key); + // 存在即命中(hash 去重):跳过重写,读侧幂等 + if !tokio::fs::try_exists(&path).await.unwrap_or(false) { + if let Err(e) = tokio::fs::write(&path, bytes).await { + // 半写残件不留盘(对齐 u-claw fetchResource 防线) + tokio::fs::remove_file(&path).await.ok(); + return Err(format!("媒体写入失败: {e}")); + } + } + match &self.oss { + Some(cfg) => match upload_to_oss(cfg, &key, bytes, media_type).await { + Ok(()) => Ok(public_url(cfg, &key)), + Err(e) => { + tracing::warn!(key = %key, err = %e, "OSS 上传失败,引用降级本机 file://"); + Ok(local_ref(&path)) + } + }, + None => Ok(local_ref(&path)), + } + } +} + +/// 内容寻址文件名:`[_<清洗原始名>]`(对齐 u-claw safeAttachName +/// 模式)——hash 前缀保去重(同内容必同文件)且排序聚簇,原始名缀后保目录可读。 +/// 平铺不分会话目录:会话归属经 DB 消息引用反查,文件系统不编码会话(否则同图跨会话 +/// 重复存储,去重失效);个人桌面应用媒体量级百~千,hash 前缀平铺可管理,分桶待量级上来再说 +fn content_key(bytes: &[u8], media_type: &str, original_name: Option<&str>) -> String { + use sha2::Digest as _; + let digest = hex::encode(&sha2::Sha256::digest(bytes)[..8]); + // 扩展名:原始名自带 ext 优先(保真),否则按 media_type 规范化 + let name = original_name.map(sanitize_name).filter(|s| !s.is_empty()); + let ext = name + .as_deref() + .and_then(|n| { + std::path::Path::new(n) + .extension() + .map(|e| format!(".{}", e.to_string_lossy())) + }) + .unwrap_or_else(|| ext_of(media_type).to_string()); + match name { + Some(n) => format!("{digest}_{n}"), + None => format!("{digest}{ext}"), + } +} + +/// 原始文件名清洗(防路径穿越/过长):/ \ .. → _,截 60 字符 +fn sanitize_name(name: &str) -> String { + let cleaned: String = name + .replace('/', "_") + .replace('\\', "_") + .replace("..", "_") + .chars() + .take(60) + .collect(); + cleaned.trim().to_string() +} + +fn ext_of(media_type: &str) -> &'static str { + match media_type { + "image/png" => ".png", + "image/jpeg" => ".jpg", + "image/gif" => ".gif", + "image/webp" => ".webp", + _ => ".bin", + } +} + +/// 本机引用(file:// + 原生绝对路径;df-ai::image_ref 展开端同前缀约定) +fn local_ref(path: &std::path::Path) -> String { + format!("file://{}", path.display()) +} + +/// OSS 公网引用:自定义 CDN 域优先,缺省桶域名 +fn public_url(cfg: &OssConfig, key: &str) -> String { + if cfg.public_base_url.is_empty() { + format!("https://{}.{}/{}", cfg.bucket, cfg.endpoint, key) + } else { + format!("{}/{}", cfg.public_base_url.trim_end_matches('/'), key) + } +} + +/// 阿里云 OSS PUT 上传(Header 签名版,无需 SDK): +/// Signature = base64(hmac-sha1(SK, "PUT\n\n{ctype}\n{date}\n/{bucket}/{key}")) +async fn upload_to_oss( + cfg: &OssConfig, + key: &str, + bytes: &[u8], + content_type: &str, +) -> Result<(), String> { + use hmac::Mac as _; + let url = format!("https://{}.{}/{}", cfg.bucket, cfg.endpoint, key); + let date = httpdate::fmt_http_date(std::time::SystemTime::now()); + let resource = format!("/{}/{}", cfg.bucket, key); + let string_to_sign = format!("PUT\n\n{content_type}\n{date}\n{resource}"); + let mut mac = hmac::Hmac::::new_from_slice(cfg.access_key_secret.as_bytes()) + .map_err(|e| format!("OSS 签名初始化失败: {e}"))?; + mac.update(string_to_sign.as_bytes()); + let sig = base64_sign(&mac.finalize().into_bytes()); + let client = reqwest::Client::new(); + let resp = client + .put(&url) + .header("Date", &date) + .header("Content-Type", content_type) + .header( + "Authorization", + format!("OSS {}:{sig}", cfg.access_key_id), + ) + .body(bytes.to_vec()) + .timeout(std::time::Duration::from_secs(60)) + .send() + .await + .map_err(|e| format!("OSS 请求失败: {e}"))?; + if resp.status().is_success() { + Ok(()) + } else { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + Err(format!("OSS 上传失败 {status}: {}", truncate_body(&body))) + } +} + +/// hmac 输出 → 标准 base64(签名编码) +fn base64_sign(digest: &[u8]) -> String { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD.encode(digest) +} + +/// 错误体截断(防 XML 错误页刷日志) +fn truncate_body(s: &str) -> &str { + match s.char_indices().nth(200) { + Some((i, _)) => &s[..i], + None => s, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn save_dedupes_by_content_and_refs_local() { + let dir = std::env::temp_dir().join(format!("df-media-test-{}", df_types::types::new_id())); + let store = MediaStore::new(dir.clone(), None); + let url1 = store.save(b"\x89PNG same", "image/png", None).await.unwrap(); + let url2 = store.save(b"\x89PNG same", "image/png", None).await.unwrap(); + let url3 = store.save(b"\x89PNG other", "image/png", None).await.unwrap(); + // 同内容同引用(hash 去重),不同内容不同引用 + assert_eq!(url1, url2); + assert_ne!(url1, url3); + assert!(url1.starts_with("file://"), "纯本地模式引用 file://, 实际 {url1}"); + let path = std::path::Path::new(&url1["file://".len()..]); + assert_eq!(tokio::fs::read(path).await.unwrap(), b"\x89PNG same"); + tokio::fs::remove_dir_all(&dir).await.ok(); + } + + #[tokio::test] + async fn original_name_kept_readable_and_sanitized() { + let dir = std::env::temp_dir().join(format!("df-media-test-{}", df_types::types::new_id())); + let store = MediaStore::new(dir.clone(), None); + let url = store + .save(b"file-bytes", "application/octet-stream", Some("../报表/最终 版本.xlsx")) + .await + .unwrap(); + let name = &url["file://".len()..]; + let fname = std::path::Path::new(name).file_name().unwrap().to_string_lossy(); + assert!(fname.contains('_'), "hash 与原始名以 _ 分隔: {fname}"); + assert!(!fname.contains("..") && !fname.contains('/'), "路径穿越字符已清洗: {fname}"); + assert!(fname.ends_with(".xlsx"), "原始名自带 ext 优先: {fname}"); + // 同内容同名 → 同文件(去重不因名字破坏) + let url2 = store + .save(b"file-bytes", "application/octet-stream", Some("../报表/最终 版本.xlsx")) + .await + .unwrap(); + assert_eq!(url, url2); + tokio::fs::remove_dir_all(&dir).await.ok(); + } + + #[test] + fn content_key_stable_and_ext_aware() { + let k1 = content_key(b"abc", "image/png", None); + let k2 = content_key(b"abc", "image/png", None); + let k3 = content_key(b"abc", "image/jpeg", None); + assert_eq!(k1, k2); + assert_ne!(k1, k3, "同字节不同 media_type 应不同文件(png/jpg 不混)"); + assert!(k1.ends_with(".png") && k1.len() == 16 + 4); + } + + #[test] + fn public_url_prefers_cdn_domain() { + let cfg = OssConfig { + bucket: "b".into(), + endpoint: "oss-cn-hangzhou.aliyuncs.com".into(), + access_key_id: "ak".into(), + access_key_secret: "sk".into(), + public_base_url: "https://cdn.example.com/".into(), + }; + assert_eq!(public_url(&cfg, "k.png"), "https://cdn.example.com/k.png"); + let cfg2 = OssConfig { public_base_url: String::new(), ..cfg }; + assert_eq!( + public_url(&cfg2, "k.png"), + "https://b.oss-cn-hangzhou.aliyuncs.com/k.png" + ); + } +}