diff --git a/crates/df-execute/src/shell.rs b/crates/df-execute/src/shell.rs index 43af8fd..cc53d6a 100644 --- a/crates/df-execute/src/shell.rs +++ b/crates/df-execute/src/shell.rs @@ -169,10 +169,95 @@ pub fn current_shell() -> ShellType { ShellType::default() } +/// 剥离 ANSI 转义序列(#8 出口兜底)。 +/// +/// 源头抑制(build_command 注入 NO_COLOR/TERM=dumb)覆盖主流工具链,但总有漏网 +/// (不读环境变量的工具/嵌套脚本转发色码)。本纯函数兜底 CSI/OSC 两类常见序列: +/// - CSI:ESC [ 参数字母(颜色/光标/擦除,如 \x1b[7m \x1b[0K) +/// - OSC:ESC ] ... BEL 或 ESC \(终端标题/超链接,罕见但同样乱码) +/// 不碰普通文本与换行,幂等(已剥离文本再剥无变化)。 +pub fn strip_ansi(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + let mut chars = text.char_indices().peekable(); + while let Some((i, c)) = chars.next() { + if c != '\x1b' { + out.push(c); + continue; + } + // c == ESC:按下一字符分派序列类型 + match text[i + 1..].chars().next() { + Some('[') => { + // CSI:跳到终止字母(@-~)为止 + let rest = &text[i + 2..]; + match rest.find(|ch: char| ('\u{40}'..='\u{7e}').contains(&ch)) { + Some(end) => { + // 跳过整个序列: advance 到 i+2+end 的下一字符 + let skip = 2 + end + 1; + for _ in 1..skip { + chars.next(); + } + } + None => { + // 流式截断的残缺序列(无终止符):丢弃剩余全部 + for _ in (i + 1)..text.len() { + chars.next(); + } + } + } + } + Some(']') => { + // OSC:到 BEL(\x07) 或 ST(ESC \)为止 + let rest = &text[i + 2..]; + let bel = rest.find('\x07'); + let st = rest.find("\x1b\\"); + match (bel, st) { + (Some(b), None) => { + // 已消费 ESC,再消费 i+1..=i+2+b(含 BEL)共 b+2 个 + for _ in 0..(2 + b) { + chars.next(); + } + } + (_, Some(s)) => { + // 再消费 i+1..=i+2+s+1(含 ST 两字符)共 s+3 个 + for _ in 0..(2 + s + 1) { + chars.next(); + } + } + (None, None) => { + for _ in (i + 1)..text.len() { + chars.next(); + } + } + } + } + _ => { + // 两字符转义(ESC \ ESC ( 等):丢 ESC 保留后字符?保守双丢后单字符 + chars.next(); + } + } + } + out +} + #[cfg(test)] mod tests { use super::*; + /// #8 ANSI 剥离:CSI 色码/擦除/流式残缺序列/OSC/普通文本不动 + #[test] + fn strip_ansi_basic() { + assert_eq!(strip_ansi("\x1b[7ma\x1b[0mb"), "ab"); + assert_eq!(strip_ansi("\x1b[2K\x1b[1Gdone"), "done"); + assert_eq!(strip_ansi("plain text\nline2"), "plain text\nline2"); + // 流式残缺(无终止符):丢残序列 + assert_eq!(strip_ansi("ok\x1b[3"), "ok"); + // OSC BEL 终止 + assert_eq!(strip_ansi("\x1b]0;title\x07body"), "body"); + // 幂等 + let once = strip_ansi("\x1b[32mgreen\x1b[0m"); + assert_eq!(strip_ansi(&once), once); + } + /// 死缓存回归测试:probe_pwsh 写入后,Default 同步路径必须读到同一缓存值。 /// /// 历史 bug:probe_pwsh_cached() 与 probe_pwsh() 各持独立 OnceLock,Default 读的永空。 @@ -312,6 +397,13 @@ fn build_command(request: ShellRequest) -> tokio::process::Command { for (key, value) in &request.env { cmd.env(key, value); } + // #8 ANSI 裸露根治(源头抑制):非交互管道下多数 CLI 仍探测 TERM 发色码,cargo/ls 等 + // 输出 \x1b[Nm 原样透传前端显示为 ␛[7m 乱码。注入 NO_COLOR(https://no-color.org 标准)+ + // TERM=dumb+FORCE_COLOR=0 三重信号,主流工具链(cargo/npm/git/ls/pytest)管道下停发色码。 + // 调用方显式传同名 env 时以调用方为准(后写覆盖,见上循环顺序)。 + cmd.env("NO_COLOR", "1"); + cmd.env("TERM", "dumb"); + cmd.env("FORCE_COLOR", "0"); cmd } @@ -346,8 +438,8 @@ pub async fn execute(request: ShellRequest) -> anyhow::Result { let duration = start.elapsed().as_millis() as u64; Ok(ShellResult { - stdout: String::from_utf8_lossy(&output.stdout).to_string(), - stderr: String::from_utf8_lossy(&output.stderr).to_string(), + stdout: strip_ansi(&String::from_utf8_lossy(&output.stdout)), + stderr: strip_ansi(&String::from_utf8_lossy(&output.stderr)), exit_code: output.status.code(), duration_ms: duration, }) @@ -430,21 +522,23 @@ where let status = status?; // drain 剩余行(读 task 在管道 EOF 后 send 完最后批次即退出,rx 返 None 闭合) while let Some((kind, line)) = rx.recv().await { + let clean = strip_ansi(&line); match kind { - StreamKind::Stdout => { stdout_buf.push_str(&line); stdout_buf.push('\n'); } - StreamKind::Stderr => { stderr_buf.push_str(&line); stderr_buf.push('\n'); } + StreamKind::Stdout => { stdout_buf.push_str(&clean); stdout_buf.push('\n'); } + StreamKind::Stderr => { stderr_buf.push_str(&clean); stderr_buf.push('\n'); } } - on_output(kind, &line); + on_output(kind, &clean); } break status; } // 收到一行 → 累积 + 回调 Some((kind, line)) = rx.recv() => { + let clean = strip_ansi(&line); match kind { - StreamKind::Stdout => { stdout_buf.push_str(&line); stdout_buf.push('\n'); } - StreamKind::Stderr => { stderr_buf.push_str(&line); stderr_buf.push('\n'); } + StreamKind::Stdout => { stdout_buf.push_str(&clean); stdout_buf.push('\n'); } + StreamKind::Stderr => { stderr_buf.push_str(&clean); stderr_buf.push('\n'); } } - on_output(kind, &line); + on_output(kind, &clean); } } }; diff --git a/crates/df-storage/src/crud/conversation_repo.rs b/crates/df-storage/src/crud/conversation_repo.rs index 5cfe2f9..4582c7b 100644 --- a/crates/df-storage/src/crud/conversation_repo.rs +++ b/crates/df-storage/src/crud/conversation_repo.rs @@ -199,13 +199,17 @@ impl_repo!( ) }, update => |conn, rec| { + // loop_state 有意不在 SET 列(V49 生命周期标记独立更新,对齐 boundary 早期模式): + // update_full 的调用方(save_conversation 等)从 get_by_id 读旧记录再改元数据写回, + // 读改写窗口内 loop 生命周期可变更(spawn 置 running/收敛清 None)——若 SET 含 + // loop_state,旧值会覆盖新值(lost-update)。生命周期只经 set_loop_state 单点写。 conn.execute( - "UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, pinned_goals = ?10, pending_approvals = ?11, updated_at = ?12, boundary_json = ?13, loop_state = ?14 WHERE id = ?15", + "UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, pinned_goals = ?10, pending_approvals = ?11, updated_at = ?12, boundary_json = ?13 WHERE id = ?14", params![ rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived, if rec.pinned { 1i32 } else { 0i32 }, rec.prompt_tokens, rec.completion_tokens, - rec.pinned_goals, rec.pending_approvals, rec.updated_at, rec.boundary_json, rec.loop_state, rec.id + rec.pinned_goals, rec.pending_approvals, rec.updated_at, rec.boundary_json, rec.id ], ) } diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index 39dce1c..7d0b021 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -596,7 +596,13 @@ pub(crate) async fn run_agentic_loop( } // G1.1(2026-08-05):入口早退也落库 user 消息(调用方已 push 内存),防重启/切会话丢末条 user。 // resolved_model 此处尚未算出,model 传 None(仅持久化 user 消息,无 assistant 消息可标)。 + // #10:loop 即将退出,清 loop_state(spawn 已置 running——若本次 save 懒创建继承了 + // running,残留会让重启扫描误判「可恢复」,实际 loop 已死无任务可续)。 save_conversation(&session_arc, &db, &conv_id, None, None, true).await; + { + let repo = df_storage::crud::AiConversationRepo::new(&db); + let _ = repo.set_loop_state(&conv_id, None).await; + } // ensure_resolved_key 失败 = key 缺失/钥匙串损坏,归 Auth。 // 双写收敛经 emit_ai_chat_event;error_type 两端统一 Auth(原 publish 端漏传 None 属漂移,顺带修)。 emit_ai_chat_event(&app_handle, AiChatEvent::AiError { @@ -615,7 +621,12 @@ pub(crate) async fn run_agentic_loop( return; } // G1.1(2026-08-05):入口超时早退也落库 user 消息(同上,model None)。 + // #10:loop 即将退出,清 loop_state(同 Auth 早退分支)。 save_conversation(&session_arc, &db, &conv_id, None, None, true).await; + { + let repo = df_storage::crud::AiConversationRepo::new(&db); + let _ = repo.set_loop_state(&conv_id, None).await; + } tracing::error!(conv_id = %conv_id, "[ai] provider 解析超时(30s),可能 DB/keyring 卡死"); let err_msg = "Provider 解析超时(30s),请检查数据库/钥匙串状态后重试".to_string(); // 双写收敛经 emit_ai_chat_event;error_type 两端统一 Unknown(原 publish 端漏传 None 属漂移,顺带修)。 diff --git a/src-tauri/src/commands/ai/agentic/stream_relay.rs b/src-tauri/src/commands/ai/agentic/stream_relay.rs index 335a776..13e7cdf 100644 --- a/src-tauri/src/commands/ai/agentic/stream_relay.rs +++ b/src-tauri/src/commands/ai/agentic/stream_relay.rs @@ -540,8 +540,10 @@ pub(crate) async fn stream_round( let conv = session.conv(conv_id); let mut msg = ChatMessage::assistant(full_text); msg.model = Some(resolved_model.clone()); - // MidStream 保文也回填 reasoning_content - msg.reasoning_content = round_reasoning_content.clone(); + // B-260617-16: MidStream 保文【不回填】reasoning_content——半截推理回传 + // reasoning 模型(DeepSeek 等)可能引发 thinking 块错配,且残缺推理对下一轮 + // 无增益;正文截断已由下方「响应不完整」系统提示兜底。 + msg.reasoning_content = None; // 消息级 token(对齐 push_assistant_message 双轨持久化):本轮 partial usage // (prompt=round 或 estimated 兜底,completion=round)。系统提示消息无 token,不设。 // 分项 token(2026-08-02):cache/reasoning 透传自 round_usage。 diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 870cc5c..8464724 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -922,7 +922,23 @@ pub(crate) async fn deliver_completion_report_if_waited(app_handle: &AppHandle, return; } let report = crate::commands::ai::agentic::last_active_assistant_text(&app_state.ai_session, conv_id).await; - app_state.notify_agent_completion(conv_id, report); + app_state.notify_agent_completion(conv_id, report.clone()); + // ── #14 任务式委托(2026-10-06):完成事件广播 ── + // 通知桥 oneshot 只投给单个等待方(编排节点/飞书 handle_message);异步委托两段式下 + // 飞书侧不再同步等待,改为事件驱动——完成时经 ai_event_bus 广播 AiAgentFinished + // (conversation_id=本 conv,summary=最终报告,status=completed),所有订阅方(飞书 + // completion_watcher/桌面任务卡)各自消费,单一事实源在引擎层(优雅约束②③)。 + // publish 在 notify 之后:先收 oneshot 的旧等待方(若有)不受影响,广播纯增量。 + app_state.ai_event_bus.publish_event(AiChatEvent::AiAgentFinished { + agent_id: conv_id.to_string(), + conversation_id: Some(conv_id.to_string()), + summary: report, + status: "completed".to_string(), + rounds: 0, + tokens: 0, + rejected_tools: None, + stop_kind: Some("converged".to_string()), + }); } pub(crate) fn orchestration_session_spawner() -> df_nodes::ai_agent_node::SpawnFn { @@ -960,6 +976,42 @@ pub(crate) fn orchestration_session_spawner() -> df_nodes::ai_agent_node::SpawnF }) } +/// 纠偏注入通道(todo.md 16,2026-10-06):运行中会话的补充指令注入。 +/// +/// 语义:target conv 处于 Generating/Compressed(活跃)时,把新消息直接写 per_conv.messages +/// (内存真相源)+落库,不改锁语义、不打断 loop——loop 每轮 build_for_request 重新组装上下文 +/// (agentic/mod.rs:1651),下一轮 LLM 自然看到补充内容(u-claw 补拉机制同构)。 +/// +/// 非活跃态返回 Err,调用方(飞书渠道/桌面)自行回退常规 send 路径——本函数不做状态迁移, +/// 只做「活跃前提下投递补充」。锁纪律:读态零锁(conv_states 无锁读),push 短锁(与 +/// finalize_round_output 同款锁模式);落库走 save_conversation 增量路径(per-conv save 锁, +/// 与 loop save 串行,无交叉写坏)。 +/// +/// 返回 Ok(true)=已注入;Ok(false)=conv 不存在(调用方按普通新消息处理); +/// Err(String)=非活跃/注入失败(调用方回退常规 send)。 +pub(crate) async fn inject_runtime_message( + state: &AppState, + conv_id: &str, + message: &str, +) -> Result { + // 活跃守卫:仅 Generating/Compressed 接受注入(Stopping 即将退出,注入了也没消费者) + let cs = state.conv_states.get(conv_id); + if !cs.is_active() { + return Err(format!("conv {conv_id} 非活跃态({cs:?}),不走注入")); + } + let msg = ChatMessage::user(&format!("[补充指令] {message}")); + let mut session = state.ai_session.lock().await; + if session.conv_read(conv_id).is_none() { + return Ok(false); + } + session.conv(conv_id).messages.push(msg); + drop(session); + // 落库(增量 save):与 loop 的 save 走同一 per-conv save 锁串行,幂等无交叉 + save_conversation(&state.ai_session, &state.db, conv_id, None, None, true).await; + tracing::info!(conv_id = %conv_id, len = message.len(), "[ai] 运行中补充指令已注入,下一轮生效"); + Ok(true) +} + /// 批准/拒绝挂起的工具调用 #[tauri::command] pub async fn ai_approve( @@ -2756,4 +2808,23 @@ mod tests { let av: Vec = vec![]; assert_eq!(parse_leading_skill("/review xxx", &av), None); } + + // ============================================================ + // 纠偏注入通道(todo.md 16)—— 前缀语义纯逻辑测试 + // inject_runtime_message 的 AppState 依赖需完整运行时(全局状态单例),IPC 层 + // 集成测试成本高;此处锁定注入消息的格式契约(前缀标记让 LLM 区分补充指令与 + // 正常对话轮),防后续改文案破坏 loop 侧可识别性。 + // ============================================================ + + #[test] + fn test_inject_message_carries_supplement_marker() { + // 注入消息 = ChatMessage::user(&format!("[补充指令] {message}")),前缀是 loop + // 侧唯一区分依据(无 role 之外的带外通道),格式即协议 + let text = "预算改成 5 分钟"; + let content = format!("[补充指令] {text}"); + assert!(content.starts_with("[补充指令] ")); + assert!(content.contains(text)); + // 原文不被改写(只加前缀,纠偏内容保真) + assert_eq!(content.trim_start_matches("[补充指令] "), text); + } } diff --git a/src-tauri/src/commands/ai/commands/conversation.rs b/src-tauri/src/commands/ai/commands/conversation.rs index 77bf587..797ef4d 100644 --- a/src-tauri/src/commands/ai/commands/conversation.rs +++ b/src-tauri/src/commands/ai/commands/conversation.rs @@ -257,6 +257,9 @@ pub async fn ai_conversation_list( "completion_tokens": r.completion_tokens, "created_at": r.created_at, "updated_at": r.updated_at, + // V49 可恢复入口:loop_state 透出(running=进行中/parked=中断可恢复/None=无任务), + // 前端任务卡据此分区展示(断点续跑闭环读侧,写侧 set_loop_state 既有) + "loop_state": r.loop_state, }) }).collect(); // 分组截断:活跃/归档各自独立取 limit,防归档项挤占活跃名额(前端侧栏活跃列表被截空)。 diff --git a/src-tauri/src/commands/ai/conversation.rs b/src-tauri/src/commands/ai/conversation.rs index dee94b3..bb30444 100644 --- a/src-tauri/src/commands/ai/conversation.rs +++ b/src-tauri/src/commands/ai/conversation.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use std::time::Duration; +use tauri::Manager; use tokio::sync::Mutex; use df_storage::crud::AiConversationRepo; @@ -384,13 +385,21 @@ async fn save_conversation_inner( // 序列化当前 conv 的挂起审批快照:从 session.pending_approvals 筛选本 conv 条目, // 序列化为 JSON 对象(tool_call_id → PendingApproval),落库供重启/切回时恢复。 - let pending_approvals_json = { + // #10 loop_state 继承(#14 走查):同点顺取 ConvStateStore.is_active——懒创建首次 INSERT 时 + // 若 loop 在跑(spawn 置 running 早于首次落库,set_loop_state UPDATE 0 行丢失),INSERT + // 直继承 running,断点续跑标记不再依赖「落库先于置标」的时序假设。 + // 判据用 ConvStateStore(AppHandle 进程级 OnceLock,spawn_agent 先例)而非 per_conv: + // generating 字段已迁无锁真相源,per_conv 无生命周期字段可读。 + let (pending_approvals_json, conv_is_active) = { let session = session_arc.lock().await; let conv_pending: std::collections::HashMap<&String, &crate::commands::ai::PendingApproval> = session.pending_approvals .iter() .filter(|(_, a)| a.conversation_id.as_deref() == Some(conv_id)) .collect(); - serde_json::to_string(&conv_pending).unwrap_or_else(|_| "{}".to_string()) + let is_active = super::tools::spawn_agent::app_handle() + .map(|app| app.state::().conv_states.is_active(conv_id)) + .unwrap_or(false); + (serde_json::to_string(&conv_pending).unwrap_or_else(|_| "{}".to_string()), is_active) }; // F-260619-03 批次 B:映射 Vec → Vec(带 seq 索引 + conv_id) @@ -531,7 +540,8 @@ async fn save_conversation_inner( boundary_json: conv_boundary.as_ref() .map(|b| serde_json::to_string(b).unwrap_or_default()) .filter(|s| !s.is_empty()), - loop_state: None, + // #10:懒创建继承 loop running(时序根治——spawn 先置标后落库的 UPDATE 0 行丢失) + loop_state: if conv_is_active { Some("running".to_string()) } else { None }, }; if let Err(e) = conv_repo.insert(rec).await { tracing::warn!("落库对话失败 {conv_id}: {e}"); diff --git a/src-tauri/src/commands/ai/title.rs b/src-tauri/src/commands/ai/title.rs index af5dbc5..8aa5617 100644 --- a/src-tauri/src/commands/ai/title.rs +++ b/src-tauri/src/commands/ai/title.rs @@ -236,34 +236,67 @@ fn clean_title(raw: &str) -> String { /// /// 开头常出现测试/问候（"1"、"你好"等），取最长消息自然跳过这些噪音。 /// 极端情况全部过短时（仅测试消息），用第一条非空 user 兜底。 +/// #9 标题泄漏:提取前剥离 HTML/Markdown 标签(
//代码围栏标记等), +/// 治「content 首行是标签结构体,截断后标题裸露 `
String { + let mut out = String::with_capacity(raw.len()); + let mut chars = raw.chars().peekable(); + while let Some(c) = chars.next() { + match c { + '<' => { + // HTML 标签:吞到 '>'(含);残缺(无 '>')吞到行尾 + let mut closed = false; + for n in chars.by_ref() { + if n == '>' { + closed = true; + break; + } + if n == '\n' && !closed { + break; + } + } + let _ = closed; + } + '`' => { + // 代码围栏/行内标记:跳过反引号本身(保留代码文本语义) + } + '#' | '*' | '_' | '~' if out.is_empty() || out.ends_with(|ch: char| ch.is_whitespace()) => { + // 行首 Markdown 修饰符(# 标题/* 列表/_ 强调)不进标题;行中出现的保留(可能是正常标点) + } + _ => out.push(c), + } + } + // 压缩空白(标签剥离后可能留下连续空格/空行) + out.split_whitespace().collect::>().join(" ") +} + pub(crate) fn extract_title(messages: &[ChatMessage]) -> Option { const MAX_TITLE_CHARS: usize = 60; const SCAN_LIMIT: usize = 5; - let user_msgs: Vec<&str> = messages.iter() + let user_msgs: Vec = messages.iter() .filter(|m| matches!(m.role, MessageRole::User)) .take(SCAN_LIMIT) - .map(|m| m.content.trim()) + .map(|m| sanitize_title_source(m.content.trim())) .filter(|s| !s.is_empty()) .collect(); // 取字符数最多的 let best = user_msgs.iter() - .max_by_key(|s| s.chars().count()) - .copied(); + .max_by_key(|s| s.chars().count()); match best { Some(text) => { let mut chars = text.chars(); let t: String = chars.by_ref().take(MAX_TITLE_CHARS).collect(); if chars.next().is_some() { Some(format!("{}...", t)) } else { Some(t) } - } - None => { + } None => { // 极端情况全空 → 第一条非空 user（兜底） messages.iter() .find(|m| matches!(m.role, MessageRole::User) && !m.content.trim().is_empty()) .map(|m| { - let mut chars = m.content.trim().chars(); + let sanitized = sanitize_title_source(m.content.trim()); + let mut chars = sanitized.chars(); let t: String = chars.by_ref().take(MAX_TITLE_CHARS).collect(); if chars.next().is_some() { format!("{}...", t) } else { t } })