From 8c0607687307492f3cadd4daa1c4e08a15b845e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Wed, 7 Oct 2026 18:32:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20B5=20=E7=BB=AD=E4=BD=9C?= =?UTF-8?q?=E5=BC=95=E6=93=8E=E5=AE=8C=E6=95=B4=E6=89=B9:=20=E8=87=AA?= =?UTF-8?q?=E9=A2=86+=E5=81=9C=E9=9D=A0=E6=81=A2=E5=A4=8D+=E7=9B=91?= =?UTF-8?q?=E6=8E=A7=E6=8E=A8=E9=80=81+=E6=8F=90=E4=BA=A4=E5=86=BB?= =?UTF-8?q?=E7=BB=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - B5 引擎: 看板轮询自动领单(G2 门禁 fail-closed, 领单原子写), parked 自动恢复优先, 收敛回读续作, 3 次熔断自停, KV 配置默认关 - 引擎配套: 提交冻结同步(df-commit-frozen), 工具结果外载过期清理, 工具失败快速记忆, 自监控快照飞书推送(df-monitor-chat-id) - 领单判据收紧: 正文以 🔴 开头才判高危(正文提及字样不误判) - 28 注入末轮竞态: 补充指令落末轮时顺延轮次预算(封顶 8) - 停靠卡继续伪失败根修: ai_continue_loop 非 active 时分流 parked 恢复内核(recover_parked_inner 单一事实源) - 实测: 写板 70 秒自动领单, 领单→执行→销账→回读→领下一单 全链闭环, 2050 测试过 --- crates/df-execute/src/commit_freeze.rs | 191 +++++ crates/df-execute/src/lib.rs | 4 +- crates/df-execute/src/shell.rs | 6 + src-tauri/src/commands/ai/agentic/mod.rs | 117 ++- src-tauri/src/commands/ai/audit/mod.rs | 10 +- src-tauri/src/commands/ai/audit/record.rs | 7 + src-tauri/src/commands/ai/commands/chat.rs | 81 +- .../src/commands/ai/commands/conversation.rs | 30 +- src-tauri/src/commands/ai/feishu_notify.rs | 33 + src-tauri/src/commands/ai/mod.rs | 18 + .../src/commands/ai/tool_failure_memory.rs | 153 ++++ .../src/commands/ai/tool_result_offload.rs | 162 ++++ src-tauri/src/commands/ai/unattended.rs | 791 ++++++++++++++++++ src-tauri/src/lib.rs | 6 + 14 files changed, 1580 insertions(+), 29 deletions(-) create mode 100644 crates/df-execute/src/commit_freeze.rs create mode 100644 src-tauri/src/commands/ai/feishu_notify.rs create mode 100644 src-tauri/src/commands/ai/tool_failure_memory.rs create mode 100644 src-tauri/src/commands/ai/tool_result_offload.rs create mode 100644 src-tauri/src/commands/ai/unattended.rs diff --git a/crates/df-execute/src/commit_freeze.rs b/crates/df-execute/src/commit_freeze.rs new file mode 100644 index 0000000..166963e --- /dev/null +++ b/crates/df-execute/src/commit_freeze.rs @@ -0,0 +1,191 @@ +//! 提交冻结机械闸(39 治理) — AI 引擎派发命令的 git commit 拦截。 +//! +//! 冻结期间(用户 🔒 令)禁止引擎/挂机会话执行 git commit:命令在 spawn 前被拒, +//! 改动留在工作区。用户本人终端不受影响(本闸只作用于引擎 execute 入口)。 +//! +//! 冻结标志解析(优先级:环境变量 DEVFLOW_COMMIT_FROZEN > env 内 DevFlow 标志): +//! - env::var("DEVFLOW_COMMIT_FROZEN")="1"/"true" → 冻结 +//! - ShellRequest.env 同名键同样生效(测试/调用方定向注入) +//! 默认不冻结(用户本人正常提交流程零影响)。 +//! +//! 拦截范围:命令子串含 `git commit`(含 `git -C x commit`)即拒;`git commit --amend` +//! 同拒。`git log/diff/status/add/push` 不拦(只冻结提交动作)。 +//! +//! 提示文案给 LLM 指明出路:改留工作区,等「攒批提交通知」。 + +use std::sync::atomic::{AtomicBool, Ordering}; + +use crate::shell::ShellRequest; + +/// 进程级冻结标志(引擎侧写入:unattended 轮询每轮按看板 🔒 行 + KV 覆盖同步)。 +/// 用静态位而非 std::env::set_var(Rust 2024 edition set_var unsafe 且多线程 UB)。 +static FROZEN: AtomicBool = AtomicBool::new(false); + +/// 引擎侧设置/解除冻结(unattended 轮询每轮调用)。 +pub fn set_frozen(v: bool) { + FROZEN.store(v, Ordering::SeqCst); +} + +/// 环境变量真值判定:"1"/"true"(含大小写变体)为冻结,其余("0"/"false"/空)为不冻结。 +fn env_val_is_truthy(v: &str) -> bool { + matches!(v, "1" | "true" | "TRUE" | "True") +} + +/// 是否启用冻结:静态标志(引擎同步) 或 环境变量 DEVFLOW_COMMIT_FROZEN(调用方/测试定向注入)。 +pub fn is_frozen(extra_env: Option<&std::collections::HashMap>) -> bool { + if FROZEN.load(Ordering::SeqCst) { + return true; + } + if let Some(v) = extra_env.and_then(|m| m.get("DEVFLOW_COMMIT_FROZEN")) { + return env_val_is_truthy(v); + } + std::env::var("DEVFLOW_COMMIT_FROZEN") + .map(|v| env_val_is_truthy(&v)) + .unwrap_or(false) +} + +/// 命令是否为 git commit 提交动作(token 判据,含 -C 前缀形态与复合命令)。 +pub fn is_commit_command(command: &str) -> bool { + // 按常见命令分隔符切段( ; && || | 换行),每段独立判 git commit(防 "x; git commit" 绕过) + command + .split(|c: char| matches!(c, ';' | '|' | '&' | '\n')) + .any(is_commit_segment) +} + +/// 单段判据:token 找 "git",跳过全局选项(-C path 等)后是否接 "commit"。 +fn is_commit_segment(segment: &str) -> bool { + let tokens: Vec = segment + .split_whitespace() + .map(|t| t.trim_matches(|c: char| c == '"' || c == '\'').to_lowercase()) + .collect(); + for (i, t) in tokens.iter().enumerate() { + if t == "git" { + let mut j = i + 1; + while j < tokens.len() { + match tokens[j].as_str() { + // `-c key=val` 两 token 形态:值在下一 token,跳 2 + "-c" => j += 2, + // `--git-dir=path`/`--work-tree=path` 单 token 合并形态:自含值,跳 1 + t if t.starts_with("--git-dir=") || t.starts_with("--work-tree=") => j += 1, + // `--git-dir path`/`--work-tree path` 两 token 形态:跳 2 + "--git-dir" | "--work-tree" => j += 2, + "commit" => return true, + _ => break, + } + } + } + } + false +} + +/// 冻结闸检查入口:冻结在场且命令是 git commit → 返回拒绝 Err(不 spawn)。 +pub fn check(request: &ShellRequest) -> Result<(), String> { + if !is_frozen(Some(&request.env)) { + return Ok(()); + } + if is_commit_command(&request.command) { + return Err(format!( + "[提交冻结令] 检测到 git commit 命令,当前处于提交冻结期(用户 🔒 令),已拒绝执行: \ + 改动保留在工作区,等用户「攒批提交通知」后才可提交。命令: {}", + truncate_cmd(&request.command) + )); + } + Ok(()) +} + +fn truncate_cmd(s: &str) -> String { + if s.chars().count() <= 120 { + s.to_string() + } else { + let cut: String = s.chars().take(120).collect(); + format!("{}…", cut) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + fn req(cmd: &str) -> ShellRequest { + ShellRequest { + command: cmd.to_string(), + working_dir: None, + env: HashMap::new(), + timeout_secs: None, + shell_type: None, + } + } + + fn req_env(cmd: &str, env: &[(&str, &str)]) -> ShellRequest { + let mut r = req(cmd); + r.env = env + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + r + } + + #[test] + fn commit_detection_covers_variants() { + assert!(is_commit_command("git commit -m 'x'")); + assert!(is_commit_command("git -C E:/repo commit -m x")); + assert!(is_commit_command("git commit --amend")); + assert!(is_commit_command("git commit -m y")); + assert!(is_commit_command("git commit; git push")); + // PowerShell / sh 包裹形态 + assert!(is_commit_command("powershell -Command git commit -m x")); + assert!(is_commit_command("cd repo && git commit -m x")); + assert!(is_commit_command("git add . && git commit -m x")); + } + + #[test] + fn non_commit_git_not_blocked() { + assert!(!is_commit_command("git log -1")); + assert!(!is_commit_command("git status")); + assert!(!is_commit_command("git diff --stat")); + assert!(!is_commit_command("git add .")); + assert!(!is_commit_command("git push origin main")); + assert!(!is_commit_command("cargo test")); + // commitment 等误命中防护:commit 后无后续 token 才算 + assert!(!is_commit_command("git log --grep commit")); + } + + #[test] + fn freeze_gate_blocks_commit_when_frozen() { + let r = req_env("git commit -m '细项批'", &[("DEVFLOW_COMMIT_FROZEN", "1")]); + assert!(check(&r).is_err(), "冻结在场 commit 必须被拒"); + let r2 = req_env("git log -1", &[("DEVFLOW_COMMIT_FROZEN", "1")]); + assert!(check(&r2).is_ok(), "冻结只拦 commit 不拦只读命令"); + } + + /// 解冻态放行用例:持与 static_flag_toggle 同一把串行闸——静态位是进程级单源, + /// 置位瞬间并行跑到本用例会误判冻结(实测互踩),凡依赖「静态位=false」断言的用例一律持闸。 + #[test] + fn no_freeze_no_block() { + let _gate = crate::shell::PWSH_CACHE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let r = req("git commit -m '用户本人提交'"); + assert!(check(&r).is_ok(), "无冻结标志零影响(用户正常提交不受影响)"); + } + + #[test] + fn frozen_false_value_no_block() { + let _gate = crate::shell::PWSH_CACHE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let r = req_env("git commit -m x", &[("DEVFLOW_COMMIT_FROZEN", "0")]); + assert!(check(&r).is_ok(), "显式关闭标志不拦(解冻态)"); + } + + /// 静态标志(引擎同步路径)切换:置位拦 commit,复位恢复。 + /// 串行闸:静态位是进程级单源,置位期间并行的 env 用例(check 走 is_frozen 静态位短路)会互踩, + /// 持锁串行 + 进出恢复默认(false)防污染。 + #[test] + fn static_flag_toggle() { + // 复用 PWSH_CACHE_TEST_LOCK 串行闸(进程级互斥,专用闸后续有需要再单立) + let _gate = crate::shell::PWSH_CACHE_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + set_frozen(true); + assert!(check(&req("git commit -m x")).is_err(), "静态冻结位拦 commit"); + assert!(check(&req("git diff")).is_ok(), "静态冻结位不拦只读命令"); + set_frozen(false); + assert!(check(&req("git commit -m x")).is_ok(), "复位后恢复放行"); + } +} diff --git a/crates/df-execute/src/lib.rs b/crates/df-execute/src/lib.rs index 6b0ec9a..90dee91 100644 --- a/crates/df-execute/src/lib.rs +++ b/crates/df-execute/src/lib.rs @@ -1,6 +1,8 @@ -//! df-execute: 执行运行时 — Shell + 环境感知 +//! df-execute: 执行运行时 — Shell + 环境感知 + 提交冻结闸 +pub mod commit_freeze; pub mod env_snapshot; pub mod shell; +pub use commit_freeze::is_commit_command; pub use env_snapshot::EnvSnapshot; diff --git a/crates/df-execute/src/shell.rs b/crates/df-execute/src/shell.rs index cc53d6a..99bfdeb 100644 --- a/crates/df-execute/src/shell.rs +++ b/crates/df-execute/src/shell.rs @@ -416,6 +416,9 @@ fn build_command(request: ShellRequest) -> tokio::process::Command { pub async fn execute(request: ShellRequest) -> anyhow::Result { let start = std::time::Instant::now(); + // 提交冻结机械闸(39 治理):冻结在场且命令为 git commit → spawn 前拒绝,改动留工作区 + crate::commit_freeze::check(&request).map_err(anyhow::Error::msg)?; + // 探测 pwsh(惰性 + OnceLock 全局缓存,只探一次),使后续 ShellType::default() 可读取缓存 #[cfg(windows)] let _ = probe_pwsh().await; @@ -464,6 +467,9 @@ where { let start = std::time::Instant::now(); + // 提交冻结机械闸(39 治理):同 execute,流式路径同样 spawn 前拦截 + crate::commit_freeze::check(&request).map_err(anyhow::Error::msg)?; + #[cfg(windows)] let _ = probe_pwsh().await; diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index 2f1220f..5ff6056 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -423,6 +423,33 @@ pub(crate) use helpers::{ DestructiveLoopSample, destructive_loop_sample, detect_destructive_loop, }; +/// 末轮注入扩轮单 loop 累计上限(todo.md 28):每次末轮注入顺延 1 轮,封顶防注入洪水把 +/// 轮次预算无限放大(每扩轮都是真实 LLM 开销);打满上限后注入仍入消息,由下一轮或收尾轮可见。 +const INJECT_EXTRA_ROUNDS_CAP: usize = 8; + +/// 末轮注入扩轮的终点判定纯函数(todo.md 28):loop 迭代到达动态轮次终点时,按注入侧顺延 +/// 余量决定「续跑几轮 or 收尾」。抽纯函数便于单测锁定「末轮注入→有续轮」契约(完整 loop +/// 依赖 Tauri 运行时,集成测试成本高;契约由本函数与注入侧 injection_hits_final_round +/// (commands/ai/commands/chat.rs)分半锁定)。调用方已保证「iteration 已达动态终点」。 +/// +/// 返回 Some(take)=消费 take 轮余量继续;None=收尾(break,等价原 for 自然结束)。 +fn inject_extra_rounds_decision( + effective_max: usize, + consumed_extra: usize, + granted: usize, +) -> Option { + // 不限轮次 loop(max_iterations=0)到不了终点,判定分支不可达;防御返回收尾 + if effective_max == usize::MAX { + return None; + } + // 余量封顶:单 loop 累计扩轮不超过 INJECT_EXTRA_ROUNDS_CAP + let room = INJECT_EXTRA_ROUNDS_CAP.saturating_sub(consumed_extra); + if granted == 0 || room == 0 { + return None; + } + Some(granted.min(room)) +} + /// Agentic 循环:流式接收 → 工具执行 → 结果回传 LLM → 循环 /// /// 退出条件: @@ -458,11 +485,16 @@ pub(crate) async fn run_agentic_loop( // conv_read 不惰性建(与旧 stop_flag/notify 取用段一致):conv 已删(per_conv 无条目) // → 旧 loop 直接退出(ConvStateStore 条目应已被 conversation_delete 同步清)。 let (stop_flag, notify, loop_epoch_arc, last_heartbeat, mut guard) = { - let session = session_arc.lock().await; - let Some(conv) = session.conv_read(&conv_id) else { + let mut session = session_arc.lock().await; + if session.conv_read(&conv_id).is_none() { tracing::warn!(stale_conv = %conv_id, "[ai] loop 入口 conv 已删,退出"); return; - }; + } + let conv = session.conv(&conv_id); + // 末轮注入扩轮基准(todo.md 28):写本 loop 的轮次预算快照(注入侧凭它判「末轮」; + // 热改 agent_max_iterations 不影响在跑 loop)+ 清零上 loop 残留的注入顺延余量。 + conv.loop_max_rounds = max_iterations; + conv.round_budget_extra = 0; ( conv.stop_flag.clone(), conv.notify.clone(), @@ -1114,6 +1146,9 @@ pub(crate) async fn run_agentic_loop( // 0 = 不限:effective_max=usize::MAX,for 到不了上界,靠 stop_flag/收敛/审批退出(下方达上限暂停分支不触发) let effective_max = if max_iterations == 0 { usize::MAX } else { max_iterations }; + // 末轮注入扩轮(todo.md 28):for 上界放宽 CAP,真实终点由轮顶动态判定(下方); + // 无注入扩轮时动态判定与原上界等价,行为零变化。扩轮总数封顶防注入洪水无限续跑。 + let mut injected_extra_rounds: usize = 0; // H2 token 三闸基点:loop 启动挂钟起点(轮间粒度检查,单轮不中断) let loop_started = tokio::time::Instant::now(); @@ -1141,7 +1176,32 @@ pub(crate) async fn run_agentic_loop( // 范围:仅 loop 主体;provider 解析/配额获取等已有独立错误分支的段落不包,避免语义重叠。 // 依赖 panic=unwind(tokio 默认),panic=abort 构建下 catch_unwind 不生效(尽力兜底)。 let outcome = unwind_to_result(async { - for iteration in start_iteration..effective_max { + for iteration in start_iteration..effective_max.saturating_add(INJECT_EXTRA_ROUNDS_CAP) { + // ── 末轮注入扩轮判定(todo.md 28):到原轮次终点时消费注入侧顺延余量,有则续跑 + // (补充指令获得带工具的完整消费轮),无则 break 与原 for 自然结束等价,落下方 + // 既有 max_rounds 收尾分支(收敛出口零改动)。判定收敛到纯函数,单测锁契约。 + if iteration >= effective_max.saturating_add(injected_extra_rounds) { + let granted = { + let mut session = session_arc.lock().await; + if session.per_conv.contains_key(&conv_id) { + let c = session.conv(&conv_id); + std::mem::take(&mut c.round_budget_extra) + } else { + 0 + } + }; + match inject_extra_rounds_decision(effective_max, injected_extra_rounds, granted) { + Some(take) => { + injected_extra_rounds += take; + tracing::info!( + conv_id = %conv_id, + extra_rounds = injected_extra_rounds, + "[ai] 末轮注入触发扩轮,轮次预算顺延 {} 轮", take + ); + } + None => break, + } + } // ── H2 三闸(B2):轮间粒度检查,超限安全停靠(轮内不中断,下一轮头生效) ── { let elapsed = loop_started.elapsed().as_secs(); @@ -1218,7 +1278,7 @@ pub(crate) async fn run_agentic_loop( emit_ai_chat_event(&app_handle, AiChatEvent::AiTaskProgress { conversation_id: Some(conv_id.clone()), round: (iteration + 1) as u32, - max_rounds: max_iterations as u32, + max_rounds: (max_iterations + injected_extra_rounds) as u32, tokens_used: tokens.total(), }); } @@ -2794,6 +2854,10 @@ pub(crate) async fn emit_ai_completed_once( let _ = repo.set_loop_state(&cid, None).await; }); } + // B5 无人值守续作引擎收敛回读钩子:引擎在途会话到达本出口即清其标志并唤醒轮询器 + // 提前扫板续作(非引擎会话到达时引擎侧比对 conv 不匹配零动作;残留兜底在引擎轮内 + // can_accept_request 判据)。本 helper 是全部正常出口统一单点,钩子挂此处即全覆盖。 + crate::commands::ai::unattended::notify_batch_done(conv_id); // emit 端(前端通道):incomplete 用 emit_incomplete(MidStream 传 Some(true))。 // token 分项(2026-08-02):cache_hit/cache_miss/reasoning 透传前端分计费展示。 // publish 端:do_publish=false(max_iterations)跳过 publish(保留原无 publish 行为); @@ -3412,3 +3476,46 @@ mod round_reentry_tests { assert_eq!(ROUND_REENTRY_BACKOFF_SECS, [60, 300, 900]); } } + +// 末轮注入扩轮(todo.md 28)loop 侧契约:到达原轮次终点时,注入侧顺延的余量必须被消费 +// 续跑(「末轮注入→有续轮」),无余量则收尾(与原 for 自然结束等价,零行为变化)。 +// 与注入侧判定 injection_hits_final_round(commands/ai/commands/chat.rs)分半锁同一契约。 +#[cfg(test)] +mod inject_extra_rounds_tests { + use super::{inject_extra_rounds_decision, INJECT_EXTRA_ROUNDS_CAP}; + + #[test] + fn final_round_with_grant_continues() { + // 末轮注入→有续轮:原终点处有余量,续 1 轮(核心契约) + assert_eq!(inject_extra_rounds_decision(10, 0, 1), Some(1)); + // 多次末轮注入余量叠加,一次全消费(上限内) + assert_eq!(inject_extra_rounds_decision(10, 0, 3), Some(3)); + } + + #[test] + fn final_round_without_grant_ends() { + // 无注入:终点处无余量,收尾(行为与原 for 自然结束逐路径等价) + assert_eq!(inject_extra_rounds_decision(10, 0, 0), None); + } + + #[test] + fn extra_rounds_capped() { + // 累计扩轮封顶 INJECT_EXTRA_ROUNDS_CAP:room 不足只消费剩余额度,打满后不再扩轮 + assert_eq!( + inject_extra_rounds_decision(10, INJECT_EXTRA_ROUNDS_CAP - 1, 3), + Some(1), + "room=1 时只消费 1 轮", + ); + assert_eq!( + inject_extra_rounds_decision(10, INJECT_EXTRA_ROUNDS_CAP, 2), + None, + "打满上限后不再扩轮", + ); + } + + #[test] + fn unlimited_loop_never_extends() { + // max_iterations=0 不限轮次:effective_max=usize::MAX,终点分支不可达,防御收尾 + assert_eq!(inject_extra_rounds_decision(usize::MAX, 0, 5), None); + } +} diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 91be415..90cc936 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -1176,10 +1176,12 @@ pub(crate) async fn process_tool_calls( let mut in_round_dupes: Vec<(ToolCallDraft, RiskLevel, String)> = Vec::new(); let mut low_risk_uncached: Vec<(ToolCallDraft, serde_json::Value, RiskLevel)> = Vec::with_capacity(low_risk.len()); for (draft, args, risk_level) in low_risk { - let args_key = readonly_cache_args_key(&draft.name, &args); - let key = (draft.name.clone(), args_key.clone()); - if !seen_round_keys.insert(key) { - // 轮内重复:延后复用首个结果(首个已在 seen/uncached/backfill),不重复历史扫描+DB 查。 + // key 与跨轮缓存同源(readonly_cache_args_key),轮内轮外两处口径一致 + let args_val = serde_json::from_str(&draft.args).unwrap_or(serde_json::Value::Null); + let args_key = readonly_cache_args_key(&draft.name, &args_val); + // AC-EFF-T1-2 轮内去重:同轮同 (tool, args_key) 只真执行一次;后续重复记入 + // in_round_dupes 延后到 backfill 后复用 in_round_cache(防双执行)。 + if !seen_round_keys.insert((draft.name.clone(), args_key.clone())) { in_round_dupes.push((draft, risk_level, args_key)); continue; } diff --git a/src-tauri/src/commands/ai/audit/record.rs b/src-tauri/src/commands/ai/audit/record.rs index c6c377d..1e3b04b 100644 --- a/src-tauri/src/commands/ai/audit/record.rs +++ b/src-tauri/src/commands/ai/audit/record.rs @@ -24,6 +24,7 @@ use crate::commands::now_millis; use crate::state::AppState; use super::risk_str; +use crate::commands::ai::tool_result_offload; /// 构造一条工具执行审计记录(纯函数,单条 [`audit_tool_call`] / 批量插入路径共用)。 /// @@ -41,6 +42,12 @@ pub(crate) fn build_audit_record( message_id: Option<&str>, ) -> AiToolExecutionRecord { let executed_at = if decided_by.is_some() { Some(now_millis()) } else { None }; + // D2a 结果外置:超 8KB 的审计 result 落盘留句柄,治 DB 膨胀(devflow.db 193.7MB 实证)。 + // 失败/未注入 base 时原样返回,降级安全。仅 completed/failed 有 result 的记录受益。 + let result = match result { + Some(r) if !r.is_empty() => Some(tool_result_offload::offload_result(tool_call_id, tool_name, &r)), + other => other, + }; AiToolExecutionRecord { id: new_id(), conversation_id: Some(conv_id.to_string()), diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 3db5601..7b6b5bb 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -1108,6 +1108,15 @@ pub(crate) fn orchestration_session_spawner() -> df_nodes::ai_agent_node::SpawnF }) } +/// 注入侧末轮判定纯函数(todo.md 28):cur_round 为 1-based 当前轮(per_conv.iteration_used, +/// prepare_round_inputs 每轮头写入),loop_max 为本 loop 轮次预算快照(per_conv.loop_max_rounds, +/// run_agentic_loop 入口写入)。loop_max=0 表示无预算概念(loop 未起/子代理会话/不限轮次) +/// → 恒 false 不扩轮。与 loop 侧 inject_extra_rounds_decision(agentic/mod.rs)分半锁定 +/// 「末轮注入→有续轮」契约。 +fn injection_hits_final_round(cur_round: usize, loop_max: usize) -> bool { + loop_max > 0 && cur_round >= loop_max +} + /// 纠偏注入通道(todo.md 16,2026-10-06):运行中会话的补充指令注入。 /// /// 语义:target conv 处于 Generating/Compressed(活跃)时,把新消息直接写 per_conv.messages @@ -1136,11 +1145,20 @@ pub(crate) async fn inject_runtime_message( if session.conv_read(conv_id).is_none() { return Ok(false); } - session.conv(conv_id).messages.push(msg); + let conv = session.conv(conv_id); + conv.messages.push(msg); + // 末轮竞态修补(todo.md 28):注入落在 loop 末轮时,消息只能被无工具收尾轮当文本消化 + // (无法执行)甚至彻底无人消费。判定命中则顺延一轮预算,loop 跑到原轮次终点时消费 + // 余量续跑一轮,补充指令获得带工具的完整消费轮。判定与顺延同锁(session lock),与 + // loop 入口快照写入/终点余量消费串行,无 lost update。 + let extended = injection_hits_final_round(conv.iteration_used, conv.loop_max_rounds); + if extended { + conv.round_budget_extra += 1; + } 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] 运行中补充指令已注入,下一轮生效"); + tracing::info!(conv_id = %conv_id, len = message.len(), extended, "[ai] 运行中补充指令已注入,下一轮生效"); Ok(true) } @@ -2662,6 +2680,7 @@ pub async fn ai_continue_loop( state: State<'_, AppState>, conversation_id: String, ) -> Result { + let mut not_active = false; { let __lock_t1744 = std::time::Instant::now(); let mut session = state.ai_session.lock().await; @@ -2670,15 +2689,36 @@ pub async fn ai_continue_loop( // B-Phase3:读侧切无锁 conv_states。 if !state.conv_states.is_active(&conversation_id) { log_slow_lock(__lock_t1744, "ai_continue_loop:1744"); - return Err("AI 未在暂停态,无需继续".to_string()); + not_active = true; + } else { + let conv = session.conv(&conversation_id); + // 复位停止信号:暂停态可能因上一轮 stop_flag 残留为 true,续跑 loop 入口会立即退出走完成流程 + conv.stop_flag.store(false, Ordering::SeqCst); + // F-260616-11 决策 a: 达 max 续跑重计 iteration(F-260616-03 决策 a,用户点继续=授权重来)。 + // 审批续跑(ai_approve)累计不重置见另路径;本路径 start_iteration 传 0,loop 从头计数。 + conv.iteration_used = 0; + log_slow_lock(__lock_t1744, "ai_continue_loop:1744"); } - let conv = session.conv(&conversation_id); - // 复位停止信号:暂停态可能因上一轮 stop_flag 残留为 true,续跑 loop 入口会立即退出走完成流程 - conv.stop_flag.store(false, Ordering::SeqCst); - // F-260616-11 决策 a: 达 max 续跑重计 iteration(F-260616-03 决策 a,用户点继续=授权重来)。 - // 审批续跑(ai_approve)累计不重置见另路径;本路径 start_iteration 传 0,loop 从头计数。 - conv.iteration_used = 0; - log_slow_lock(__lock_t1744, "ai_continue_loop:1744"); + } + if not_active { + // 非活跃分两种: 持久层 loop_state=parked(安全停靠家族,停靠时内存态已复位)走正典恢复链; + // 其余才是真无需继续。停靠卡「继续」按钮从此不再撞「未在暂停态」伪失败。 + if let Some(record) = state + .ai_conversations + .get_by_id(&conversation_id) + .await + .map_err(|e| e.to_string())? + { + if record.loop_state.as_deref() == Some("parked") { + return super::conversation::recover_parked_inner( + &app, + state.inner(), + &conversation_id, + ) + .await; + } + } + return Err("AI 未在暂停态,无需继续".to_string()); } // 复用审批恢复续 loop 入口(不重写 loop),其内部 spawn run_agentic_loop // F-09 B 批2:传 conv_id(IPC 参数 conversation_id)。 @@ -2975,4 +3015,25 @@ mod tests { // 原文不被改写(只加前缀,纠偏内容保真) assert_eq!(content.trim_start_matches("[补充指令] "), text); } + + // ============================================================ + // 末轮注入扩轮(todo.md 28)—— 注入侧末轮判定纯逻辑测试 + // inject_runtime_message 的 AppState 依赖需完整运行时,契约与 loop 侧 + // inject_extra_rounds_decision(agentic/mod.rs)分半锁定:注入落在末轮 + // (iteration_used 达 loop_max_rounds 快照)必须判末轮并顺延预算, + // 非末轮/无快照不扩轮(零行为变化)。 + // ============================================================ + + #[test] + fn test_injection_final_round_detection() { + // 末轮:当前轮(1-based)达到 loop 轮次预算快照 → 判末轮,触发顺延 + assert!(injection_hits_final_round(10, 10)); + // 读数超前(审批续跑累计后理论边界外)也判末轮,宁多扩不少丢消息 + assert!(injection_hits_final_round(11, 10)); + // 非末轮:预算未打满,不扩轮(消息由下一轮自然消费) + assert!(!injection_hits_final_round(9, 10)); + // 无预算概念(loop 未起/子代理会话/不限轮次写 0):恒不扩轮 + assert!(!injection_hits_final_round(10, 0)); + assert!(!injection_hits_final_round(0, 0)); + } } diff --git a/src-tauri/src/commands/ai/commands/conversation.rs b/src-tauri/src/commands/ai/commands/conversation.rs index d1d1f8a..ed2ccad 100644 --- a/src-tauri/src/commands/ai/commands/conversation.rs +++ b/src-tauri/src/commands/ai/commands/conversation.rs @@ -310,19 +310,33 @@ pub async fn ai_list_recoverable( /// (既有逻辑),parked 态自然清除。恢复前守卫: /// ① 会话须存在且 loop_state='parked'(非 parked 拒绝,防误触跑会话); /// ② conv_states.can_accept_request(目标 conv 真活跃则拒绝,双保险)。 +/// 恢复引导消息:LLM 上下文里既有对话历史完整(消息拆分存储回放),只需一句指令触发续跑。 +/// B5 引擎 parked 自动恢复(unattended.rs)复用同款文案,单一来源。 +pub(crate) const PARKED_RECOVERY_MESSAGE: &str = "[系统恢复] 应用重启导致上次任务中断,已完成进度已保留。请回顾上方对话上下文,继续完成未竟的任务;若上下文不足以判断进度,先向用户确认已完成的步骤再继续。"; + #[tauri::command] pub async fn ai_recover_parked( app: AppHandle, state: State<'_, AppState>, conversation_id: String, +) -> Result { + recover_parked_inner(&app, state.inner(), &conversation_id).await +} + +/// parked 恢复内核: IPC(ai_recover_parked)与桌面停靠卡「继续」(ai_continue_loop 的非活跃分流) +/// 共用的单一事实源。判据走持久层 loop_state,与会话内存态复位解耦(停靠时内存态已复位)。 +pub(crate) async fn recover_parked_inner( + app: &AppHandle, + state: &AppState, + conversation_id: &str, ) -> Result { use super::chat::spawn_agentic_session; let record = state .ai_conversations - .get_by_id(&conversation_id) + .get_by_id(conversation_id) .await - .map_err(err_str)? + .map_err(|e| e.to_string())? .ok_or_else(|| format!("对话不存在: {}", conversation_id))?; if record.loop_state.as_deref() != Some("parked") { return Err(format!( @@ -330,20 +344,18 @@ pub async fn ai_recover_parked( conversation_id, record.loop_state )); } - if !state.conv_states.can_accept_request(&conversation_id) { + if !state.conv_states.can_accept_request(conversation_id) { return Err("该会话正在生成中,无需恢复".to_string()); } - // 恢复引导消息:LLM 上下文里既有对话历史完整(消息拆分存储回放),只需一句指令触发续跑。 - let message = "[系统恢复] 应用重启导致上次任务中断,已完成进度已保留。请回顾上方对话上下文,继续完成未竟的任务;若上下文不足以判断进度,先向用户确认已完成的步骤再继续。" - .to_string(); + let message = PARKED_RECOVERY_MESSAGE.to_string(); spawn_agentic_session( - &app, - &state, + app, + state, message, None, // language:沿用会话既有 agent_language None, // skill:恢复不挂技能 None, // model_override:沿用会话既有模型 - Some(conversation_id.clone()), + Some(conversation_id.to_string()), None, // parts:纯文本 None, // mention_spans true, // pin_active:恢复动作即用户当前意图 diff --git a/src-tauri/src/commands/ai/feishu_notify.rs b/src-tauri/src/commands/ai/feishu_notify.rs new file mode 100644 index 0000000..3d79f51 --- /dev/null +++ b/src-tauri/src/commands/ai/feishu_notify.rs @@ -0,0 +1,33 @@ +//! 飞书告警出口(todo 37 监控轮询):挂机引擎自监控快照的推送通道。 +//! +//! 职责单一:把 unattended.rs run_self_monitor 的异常快照送达指定飞书群。 +//! 复用 feishu_channel 的卡片发送链路(md_to_lark_friendly 净化 + markdown 卡), +//! 目标 chat 显式传入(KV df-monitor-chat-id),不走 conv 绑定路由——监控告警 +//! 是运维通道,与会话投影(卡片滚动句柄/终态判据)互不污染。 +//! +//! 失败语义:告警是锦上添花,发送失败只落 feishu_log 不 panic 不重试—— +//! 下一监控周期同段异常仍会再次尝试(同段只告警一次的闸在调用方)。 + +use crate::feishu_channel::{feishu_log, send_markdown_card}; + +/// 监控告警推送:快照文本直发 monitor_chat(飞书 markdown 卡)。 +/// 渠道未启动/发送失败均静默降级为 feishu_log 落盘(告警不阻塞引擎主循环)。 +pub(crate) async fn push_monitor_alert(app: &tauri::AppHandle, chat_id: &str, snapshot: &str) { + let _ = app; // AppHandle 预留(将来查 DB 富化告警),当前快照自包含 + if chat_id.is_empty() { + feishu_log(&format!("[monitor] 未配置 df-monitor-chat-id,告警仅落日志: {snapshot}")); + return; + } + let Some(client) = crate::feishu_channel::active_client() else { + feishu_log(&format!( + "[monitor] 飞书渠道未启动,告警仅落日志: {snapshot}" + )); + return; + }; + let text = format!("🛡 **DevFlow 挂机监控告警**\n{snapshot}"); + let sent = send_markdown_card(client, "", chat_id, &text).await; + match sent { + Some(mid) => feishu_log(&format!("[monitor] 告警已推送 chat={chat_id} card={mid}")), + None => feishu_log(&format!("[monitor] 告警推送失败(卡片与文本均不可达),快照: {snapshot}")), + } +} diff --git a/src-tauri/src/commands/ai/mod.rs b/src-tauri/src/commands/ai/mod.rs index 0f4c5a0..7cfca05 100644 --- a/src-tauri/src/commands/ai/mod.rs +++ b/src-tauri/src/commands/ai/mod.rs @@ -48,6 +48,7 @@ pub mod knowledge_inject; pub mod lease; // 系统通知(P1 bg 化补完):完成/失败/需介入三类事件,主窗托盘化时补位提醒(条件发送) pub(crate) mod notify; +pub(crate) mod feishu_notify; pub mod prompt; pub mod provider_pool; pub mod remote_bridge; @@ -56,9 +57,15 @@ pub mod skills; pub mod stream_recv; pub mod title; pub mod tool_registry; +// B5d 工具失败记忆:同调用连败拦截+换路引导(工具学习能力,治 404 连环死磕) +pub mod tool_failure_memory; +// D2a 工具结果外置 + D2d 90 天归档:审计大结果落盘留句柄,治 DB 膨胀 +pub mod tool_result_offload; // 声明式工具注册集合(tool_registry 拆分第一步 · 试点容器)。 // 试点:list_projects 已迁声明式 declare_tool! 宏,其余 47 工具仍在 tool_registry.rs。 pub mod tools; +// B5 无人值守续作引擎:看板轮询自动领单 + parked 自动恢复(收敛回读钩子在 agentic 出口) +pub mod unattended; use agentic::conv_state::{ConvState, ConvStateStore}; @@ -1030,6 +1037,15 @@ pub struct PerConvState { pub last_heartbeat: Arc, /// agentic loop 累计已跑 iteration 计数(语义同 AiSession.iteration_used) pub iteration_used: usize, + /// 本 loop 的轮次预算快照(todo.md 28 末轮注入扩轮):run_agentic_loop 入口写入自身的 + /// max_iterations,inject_runtime_message 读它判「注入是否落在末轮」。0=无预算概念 + /// (loop 未起/子代理会话/max_iterations=0 不限轮次)→ 注入侧恒不扩轮。 + /// plain usize 同锁读写(session lock 串行),无需原子;随会话销毁不落库。 + pub loop_max_rounds: usize, + /// 注入侧顺延的轮次余量(todo.md 28):末轮注入时 inject_runtime_message +1,loop 跑到 + /// 原轮次终点时读后清零消费、续跑一轮,让补充指令获得带工具的完整消费轮。 + /// 同锁读写(session lock 串行);loop 入口清零防跨 loop 残留。 + pub round_budget_extra: usize, /// 当前 agent 循环的语言设置(用于审批后恢复循环) pub agent_language: Option, /// 用户指定模型 override(主对话专用,语义同 AiSession.model_override) @@ -1228,6 +1244,8 @@ impl PerConvState { loop_epoch: Arc::new(std::sync::atomic::AtomicU64::new(0)), last_heartbeat: Arc::new(std::sync::atomic::AtomicU64::new(0)), iteration_used: 0, + loop_max_rounds: 0, + round_budget_extra: 0, agent_language: None, model_override: None, session_trust: HashSet::new(), diff --git a/src-tauri/src/commands/ai/tool_failure_memory.rs b/src-tauri/src/commands/ai/tool_failure_memory.rs new file mode 100644 index 0000000..01d0929 --- /dev/null +++ b/src-tauri/src/commands/ai/tool_failure_memory.rs @@ -0,0 +1,153 @@ +//! B5d 工具失败记忆(conv 级):治「404 连环十几次无退避换路」——AI 工具是原生能力, +//! 应有学习能力:同 (conv, tool, args) 连败 ≥2 次 → 拦截重试,回填结构化失败提示, +//! 引导 LLM 换路(改参数/换工具/报告放弃);成功即清零(知识保鲜,防误杀)。 +//! +//! 设计要点(与既有机制的关系): +//! - 与 fail_window/熔断(stall_detector)互补:熔断治「会话级停滞」(连续 N 轮全失败), +//! 本模块治「单点死磕」(同一调用原样重发);两层独立计数互不干扰。 +//! - 与只读缓存(find_cached_readonly_result)不同:缓存复用成功结果,本模块拦截失败重复。 +//! - 记忆生命周期 = conv 生命周期(进程内 HashMap,conv 结束随会话对象丢弃;不落库, +//! 下次会话重新学习——失败参数往往任务态相关,跨会话记忆弊大于利)。 +//! - key 归一:tool_name + args 关键字段(read_file 用 path,grep 用 pattern+path, +//! run_command 用命令行本体;其余工具全 args JSON)——LLM 微调无关字段(如 limit) +//! 不应逃脱记忆,但完全相同调用必须命中。 + +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; + +/// 连败拦截阈值:同 key 连败 ≥2 次(即第 3 次尝试)在执行前拦截。 +pub const ABORT_AFTER_CONSECUTIVE_FAILURES: u32 = 2; + +/// 记忆条目:连续失败次数 + 最近一次错误摘要(拦截时回填给 LLM 换路提示)。 +#[derive(Debug, Clone)] +struct FailureEntry { + consecutive: u32, + last_error: String, +} + +/// 进程级失败记忆(全 conv 共享单锁;锁内零 await,微秒级)。 +static FAILURE_MEMORY: OnceLock>> = OnceLock::new(); + +fn memory() -> &'static Mutex> { + FAILURE_MEMORY.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// args → 记忆 key 的归一化提取(工具感知)。 +/// +/// 原则:抽「语义主键」而非全 args——read_file 微调 limit 重读同一文件,应视为同调用; +/// run_command 命令行本体即主键(含路径/参数)。 +fn memory_key(tool: &str, args: &serde_json::Value) -> String { + let get = |k: &str| args.get(k).and_then(|v| v.as_str()).unwrap_or("").to_string(); + match tool { + "read_file" | "read_symbol" | "file_info" | "delete_file" => { + format!("{tool}|path={}", get("path")) + } + "grep" => format!("{tool}|pat={}|path={}", get("pattern"), get("path")), + "search_files" => format!("{tool}|pat={}|path={}", get("pattern"), get("path")), + "list_directory" => format!("{tool}|path={}", get("path")), + "run_command" => format!("{tool}|cmd={}", get("command")), + "mysql_query" | "mysql_execute" => format!("{tool}|sql={}", get("sql")), + _ => { + // 通用:全 args 稳定序列化(serde_json Value 保序不可保证 → BTreeMap 排序归一) + let sorted = serde_json::to_string(&sort_json(args)).unwrap_or_default(); + format!("{tool}|{sorted}") + } + } +} + +/// serde_json Value 递归排序(Object 按 key 排序),保证 key 稳定。 +fn sort_json(v: &serde_json::Value) -> serde_json::Value { + use serde_json::Value; + match v { + Value::Object(m) => { + let mut sorted: Vec<(String, Value)> = + m.iter().map(|(k, v)| (k.clone(), sort_json(v))).collect(); + sorted.sort_by(|a, b| a.0.cmp(&b.0)); + Value::Object(sorted.into_iter().collect()) + } + Value::Array(a) => Value::Array(a.iter().map(sort_json).collect()), + other => other.clone(), + } +} + +/// 执行前检查:同 key 连败达标 → 拦截,返回结构化失败提示(LLM 可读换路引导)。 +/// 未达标/无记忆 → None 放行。 +pub fn precheck(tool: &str, args: &serde_json::Value) -> Option { + let key = memory_key(tool, args); + let mem = memory().lock().unwrap_or_else(|p| p.into_inner()); + let Some(entry) = mem.get(&key) else { return None }; + if entry.consecutive >= ABORT_AFTER_CONSECUTIVE_FAILURES { + Some(format!( + "{{\"status\":\"blocked_by_failure_memory\",\"tool\":\"{tool}\",\"consecutive_failures\":{},\"last_error\":\"{}\",\"hint\":\"同一调用已连败 {} 次,已自动拦截。请勿原样重试:①检查参数(路径/SQL/命令拼写) ②换工具或换路径 ③若确认无法绕过,向用户报告放弃原因。\"}}", + entry.consecutive, + entry.last_error.replace('\"', "'"), + entry.consecutive + )) + } else { + None + } +} + +/// 执行后回写:成功清零删除,失败递增并记最近错误。 +pub fn record(tool: &str, args: &serde_json::Value, ok: bool, error: Option) { + let key = memory_key(tool, args); + let mut mem = memory().lock().unwrap_or_else(|p| p.into_inner()); + if ok { + mem.remove(&key); + } else { + let e = mem.entry(key).or_insert(FailureEntry { consecutive: 0, last_error: String::new() }); + e.consecutive += 1; + e.last_error = error.unwrap_or_default().chars().take(200).collect(); + } +} + +/// 拦截条目数(测试用)。 +pub fn memory_size() -> usize { + memory().lock().unwrap_or_else(|p| p.into_inner()).len() +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn clear() { + memory().lock().unwrap().clear(); + } + + #[test] + fn precheck_blocks_after_two_failures() { + clear(); + let args = json!({"path": "/tmp/x.rs"}); + assert!(precheck("read_file", &args).is_none()); + record("read_file", &args, false, Some("文件不存在".into())); + assert!(precheck("read_file", &args).is_none()); // 1 次未达阈值 + record("read_file", &args, false, Some("文件不存在".into())); + let blocked = precheck("read_file", &args).unwrap(); + assert!(blocked.contains("blocked_by_failure_memory")); + assert!(blocked.contains("consecutive_failures\":2")); + // 无关字段微调不逃脱记忆:limit 改了仍拦截(同 path 主键) + let args2 = json!({"path": "/tmp/x.rs", "limit": 300}); + assert!(precheck("read_file", &args2).is_some()); + // 成功清零 + record("read_file", &args, true, None); + assert!(precheck("read_file", &args).is_none()); + } + + #[test] + fn memory_key_normalizes_irrelevant_fields() { + let a = json!({"path": "/a.rs", "limit": 10}); + let b = json!({"path": "/a.rs", "limit": 999}); + assert_eq!(memory_key("read_file", &a), memory_key("read_file", &b)); + // 通用工具:全 args 排序归一,键序无关 + let c = json!({"a": 1, "b": 2}); + let d = serde_json::from_str::(r#"{"b":2,"a":1}"#).unwrap(); + assert_eq!(memory_key("generic_tool", &c), memory_key("generic_tool", &d)); + } + + #[test] + fn run_command_key_is_command_itself() { + let a = json!({"command": "cargo test", "working_dir": "E:/x"}); + assert!(memory_key("run_command", &a).contains("cmd=cargo test")); + } +} diff --git a/src-tauri/src/commands/ai/tool_result_offload.rs b/src-tauri/src/commands/ai/tool_result_offload.rs new file mode 100644 index 0000000..fcd5862 --- /dev/null +++ b/src-tauri/src/commands/ai/tool_result_offload.rs @@ -0,0 +1,162 @@ +//! D2a 工具结果过期外置 + D2d 90 天归档:治 devflow.db 膨胀(193.7MB,审计大结果为主要头)。 +//! +//! 设计(对齐 u-claw 成熟实践「大内容外置+摘要留指针」): +//! - 超阈值(8KB)的审计 result 正文落盘至 {app_data}/tool-results/{yyyy-MM}/{id}.txt, +//! DB 只留句柄 JSON(offloaded 标记 + 路径 + 头部摘要 + 大小 + 工具名/时间)。 +//! - 句柄可读:句柄含 path,用户/agent 需要细节时按路径找回(工具结果在 messages 侧 +//! 已有 namespace 外置,本模块治审计表侧——两处同构,互不影响)。 +//! - D2d 归档:cleanup_expired(90 天)删除过期外置文件(句柄留 DB,文件没了读时 +//! 降级提示「已过期清理」);调用点挂 unattended 引擎每日一轮。 +//! - 历史数据默认保留禁清理(全局规则)——cleanup 只删「本模块自己写的外置文件」 +//! (目录专有,含 marker 头校验),不碰任何 DB 行,原始库零风险。 + +use std::path::PathBuf; +use std::sync::OnceLock; + +/// 外置根目录(进程级一次注入;lib.rs setup 调 init_base)。未注入时 offload_result 直通。 +static BASE: OnceLock = OnceLock::new(); + +/// 注入外置根目录(app data dir)。幂等,首次生效。 +pub fn init_base(base: PathBuf) { + let _ = BASE.set(base); +} + +/// 无参包装:build_audit_record 等审计写路径统一走此入口。 +/// base 未注入或失败降级时原样返回(不外置不丢数据)。 +pub fn offload_result(id: &str, tool_name: &str, result: &str) -> String { + match BASE.get() { + Some(b) => offload_if_large(b, id, tool_name, result), + None => result.to_string(), + } +} + +/// 外置阈值:审计 result 超 8KB 落盘(DB 只留句柄)。namespace 侧阈值 2048B 更激进 +/// (上下文敏感),审计侧存证要求低,8KB 平衡 IO 频率与瘦身收益。 +pub const OFFLOAD_THRESHOLD_BYTES: usize = 8 * 1024; + +/// D2d 归档阈值(定案 2026-10-06):外置文件 90 天后可清理。 +pub const ARCHIVE_AFTER_DAYS: u64 = 90; + +/// 句柄 JSON 字段名(读取降级判据)。 +const HANDLE_MARK: &str = "__offloaded__"; + +/// 计算外置目录({base}/tool-results/{yyyy-MM}/)。base 由调用方传入(app data dir), +/// 本模块不做 Tauri 依赖(纯函数可测)。 +fn month_dir(base: &PathBuf, id: &str) -> PathBuf { + // id 前缀含时间成分则用之,否则统一放 unsorted/;目录按月仅为人读友好 + let month = id.get(0..7).filter(|s| s.len() == 7 && s.as_bytes()[4] == b'-').unwrap_or("unsorted"); + base.join("tool-results").join(month) +} + +/// 落盘并生成句柄。失败(磁盘/权限)时返回原内容(降级:不外置也不丢数据)。 +pub fn offload_if_large(base: &PathBuf, id: &str, tool_name: &str, result: &str) -> String { + if result.len() <= OFFLOAD_THRESHOLD_BYTES { + return result.to_string(); + } + let dir = month_dir(base, id); + if std::fs::create_dir_all(&dir).is_err() { + return result.to_string(); + } + let file = dir.join(format!("{id}.txt")); + if std::fs::write(&file, result).is_err() { + return result.to_string(); + } + let head: String = result.chars().take(200).collect(); + // meta 旁路文件:cleanup_expired 的认领判据(防误删同目录非本模块产物) + let meta = serde_json::json!({ HANDLE_MARK: true, "id": id, "tool": tool_name }); + let _ = std::fs::write(file.with_extension("meta.json"), meta.to_string()); + let handle = serde_json::json!({ + HANDLE_MARK: true, + "path": file.to_string_lossy(), + "size_bytes": result.len(), + "tool": tool_name, + "head": head, + "hint": "结果过大已外置落盘,需要细节时按 path 读取原文" + }); + handle.to_string() +} + +/// 读外置文件(前端/查询侧展开句柄用)。过期清理后降级提示。 +pub fn read_offloaded(path: &str) -> Result { + std::fs::read_to_string(path).map_err(|e| format!("外置结果读取失败(可能已过 90 天归档清理): {e}")) +} + +/// D2d:清理超期外置文件(仅本模块目录内带 marker 头的文件;失败静默记日志)。 +/// 返回 (删除数, 释放字节)。全局规则「历史数据默认保留」的对齐说明:本函数只删 +/// offload 自产的派生文件(原文曾入 DB 后被替换为句柄),非用户原始数据。 +pub fn cleanup_expired(base: &PathBuf) -> (usize, u64) { + let root = base.join("tool-results"); + let Ok(entries) = std::fs::read_dir(&root) else { return (0, 0) }; + let cutoff = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) + .saturating_sub(ARCHIVE_AFTER_DAYS * 86400); + let mut removed = 0usize; + let mut freed = 0u64; + for month in entries.flatten() { + let Ok(files) = std::fs::read_dir(month.path()) else { continue }; + for f in files.flatten() { + let p = f.path(); + if p.extension().and_then(|e| e.to_str()) != Some("txt") { + continue; + } + // marker 头校验:句柄 JSON 写入文件首行注释不行(txt 是原文)——改判据: + // 同名 .meta.json 存在且 offloaded=true 才认领。写入侧同步产出 meta。 + let meta = p.with_extension("meta.json"); + let Ok(meta_str) = std::fs::read_to_string(&meta) else { continue }; + let Ok(v) = serde_json::from_str::(&meta_str) else { continue }; + if !v.get(HANDLE_MARK).and_then(|b| b.as_bool()).unwrap_or(false) { + continue; + } + let Ok(md) = f.metadata() else { continue }; + let Ok(modified) = md.modified() else { continue }; + let mtime = modified + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + if mtime < cutoff { + let size = md.len(); + if std::fs::remove_file(&p).is_ok() { + let _ = std::fs::remove_file(&meta); + removed += 1; + freed += size; + } + } + } + } + (removed, freed) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn small_result_not_offloaded() { + let base = std::env::temp_dir().join(format!("d2a-test-{}", std::process::id())); + let out = offload_if_large(&base, "2026-10-07-x", "read_file", "short"); + assert_eq!(out, "short"); + let _ = std::fs::remove_dir_all(base.join("tool-results")); + } + + #[test] + fn large_result_offloaded_with_handle() { + let base = std::env::temp_dir().join(format!("d2a-test-{}", std::process::id())); + let big = "x".repeat(10 * 1024); + let out = offload_if_large(&base, "2026-10-07-y", "run_command", &big); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(v.get(HANDLE_MARK).and_then(|b| b.as_bool()), Some(true)); + let path = v.get("path").and_then(|p| p.as_str()).unwrap(); + // 原文可找回 + assert_eq!(read_offloaded(path).unwrap(), big); + let _ = std::fs::remove_dir_all(base.join("tool-results")); + } + + #[test] + fn month_dir_parses_id_prefix() { + let base = PathBuf::from("/tmp"); + assert!(month_dir(&base, "2026-10-07-abcd").ends_with("tool-results/2026-10")); + assert!(month_dir(&base, "weird").ends_with("tool-results/unsorted")); + } +} diff --git a/src-tauri/src/commands/ai/unattended.rs b/src-tauri/src/commands/ai/unattended.rs new file mode 100644 index 0000000..7d03c8a --- /dev/null +++ b/src-tauri/src/commands/ai/unattended.rs @@ -0,0 +1,791 @@ +//! B5 无人值守续作引擎 v1(B5b+31 合并体) +//! +//! 看板(docs/todo.md 编号条目)轮询 → G2 门禁领单 → spawn_agentic_session 正典链路开工 +//! → 收敛回读 → parked 会话自动恢复,循环续作。设计定稿按稿实施,不扩面。 +//! +//! 组成: +//! - 看板解析器(纯函数):编号条目 `N. [ ]/→/x/! 正文` 解析 + 🔴 判据 +//! - 领单选择器(纯函数):G2 门禁代码复刻——存在未清零 🔴 时只领 🔴,无 🔴 按配置范围领 +//! - 轮询调度器:tokio::spawn 常驻循环,周期/开关/范围/看板路径均从 Settings KV 热读 +//! - 收敛回读钩子:agentic 收敛出口(emit_ai_completed_once)调 [`notify_batch_done`] +//! 清在途标志并唤醒轮询器(不强实时,轮询周期即延迟上界) +//! - 连续 spawn 失败熔断:3 次引擎自停(tracing error),重置=开关关→开 +//! +//! 硬约束落地: +//! - 单会话单在途:引擎内部 in_flight 标志 + 会话页 can_accept_request 双保险; +//! 标志残留(异常退出漏钩子)时会话实际空闲即自动清除,引擎不因泄漏永久卡死 +//! - 引擎任务走独立后台会话:领单传新 conv_id + pin_active=false,不劫持用户活跃面板 +//! - 看板写入只走原子写(临时文件 + rename);只改状态标记,不新增编号(G8 纪律) +//! +//! 配置面(Settings KV,通用 settings_get/settings_set IPC 可读写): +//! - `df-unattended-enabled` bool("true"/"1" 为开),默认 false +//! - `df-unattended-poll-secs` u64,默认 300(下限钳 30,防误配热循环) +//! - `df-unattended-claim-scope` "red_only"(默认)| "all_open" +//! - `df-unattended-todo-path` 看板绝对路径,默认 DevFlow 仓库根 docs/todo.md + +use std::path::Path; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::{Mutex, OnceLock}; +use std::time::Duration; + +use tauri::{AppHandle, Manager}; + +use crate::commands::ai::commands::chat::spawn_agentic_session; +use crate::state::AppState; + +// ============================================================ +// 配置 +// ============================================================ + +const KV_ENABLED: &str = "df-unattended-enabled"; +const KV_POLL_SECS: &str = "df-unattended-poll-secs"; +const KV_CLAIM_SCOPE: &str = "df-unattended-claim-scope"; +const KV_TODO_PATH: &str = "df-unattended-todo-path"; +/// 提交冻结 KV 标志(39 治理):Some(true)=强制冻结/Some(false)=强制解冻(覆盖看板判据) +const KV_COMMIT_FROZEN: &str = "df-commit-frozen"; +/// 37 监控轮询:告警/心跳摘要推送目标(飞书 chat_id)。空=不推飞书,只落 tracing+feishu_log。 +/// todo 37 验收「静默/进程消失 1 个周期内被发现并通报」的通报出口 +const KV_MONITOR_CHAT: &str = "df-monitor-chat-id"; + +/// 看板 🔒 冻结行判据(39 治理):看板任意行(trim 后)以「🔒」开头即视为冻结令生效。 +/// 兼容 blockquote 形态 `> 🔒 ...`(现行看板冻结令即此形态);正文提及「🔒」但非行首不误判。 +/// 纯函数,随每轮读板同步判定;KV `df-commit-frozen` 显式 true/false 可覆盖看板判据。 +fn board_frozen(content: &str) -> bool { + content.lines().any(|l| { + l.trim_start() + .trim_start_matches('>') + .trim_start() + .starts_with("🔒") + }) +} + +/// 冻结终判:KV 显式标志(字符串 "true"/"1" 冻结,"false"/"0" 解冻)优先,否则看板 🔒 行。 +fn resolve_frozen(board_content: Option<&str>, kv_get: &dyn Fn(&str) -> Option) -> bool { + if let Some(v) = kv_get(KV_COMMIT_FROZEN) { + let v = v.trim().trim_matches('"').to_lowercase(); + if v == "true" || v == "1" { + return true; + } + if v == "false" || v == "0" { + return false; + } + } + board_content.map(board_frozen).unwrap_or(false) +} + +/// 看板默认路径(Tauri 打包后 cwd 不定,引擎不猜 cwd,默认锚定仓库根;可 KV 覆盖) +const DEFAULT_TODO_PATH: &str = "E:\\wk-lab\\DevFlow\\docs\\todo.md"; +const DEFAULT_POLL_SECS: u64 = 300; +const MIN_POLL_SECS: u64 = 30; +/// 连续 spawn 失败熔断阈值(设计定稿:3 次) +const MAX_CONSECUTIVE_FAILURES: u32 = 3; + +/// 领单范围(claim_scope) +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClaimScope { + /// 只领 🔴 条目(G2 门禁严格档,默认) + RedOnly, + /// 无 🔴 未清零时领任意 `[ ]` 条目(🔴 在场时仍被 G2 门禁压住只领 🔴) + AllOpen, +} + +impl ClaimScope { + fn from_str(s: &str) -> Self { + match s.trim() { + "all_open" => ClaimScope::AllOpen, + _ => ClaimScope::RedOnly, + } + } +} + +/// 引擎运行期配置(每轮从 KV 热读,改配置无需重启) +#[derive(Debug, Clone, PartialEq, Eq)] +struct EngineConfig { + enabled: bool, + poll_secs: u64, + claim_scope: ClaimScope, + todo_path: String, + /// 37 监控轮询:告警/心跳摘要推送目标(飞书 chat_id),None=只落日志 + monitor_chat: Option, +} + +impl Default for EngineConfig { + fn default() -> Self { + EngineConfig { + enabled: false, + poll_secs: DEFAULT_POLL_SECS, + claim_scope: ClaimScope::RedOnly, + todo_path: DEFAULT_TODO_PATH.to_string(), + monitor_chat: None, + } + } +} + +impl EngineConfig { + fn from_kv(get: &dyn Fn(&str) -> Option) -> Self { + let mut cfg = EngineConfig::default(); + if let Some(v) = get(KV_ENABLED) { + let v = v.trim().trim_matches('"').to_lowercase(); + cfg.enabled = v == "true" || v == "1"; + } + if let Some(v) = get(KV_POLL_SECS) { + if let Ok(secs) = v.trim().trim_matches('"').parse::() { + cfg.poll_secs = secs.max(MIN_POLL_SECS); + } + } + if let Some(v) = get(KV_CLAIM_SCOPE) { + cfg.claim_scope = ClaimScope::from_str(&v); + } + if let Some(v) = get(KV_TODO_PATH) { + let v = v.trim().trim_matches('"'); + if !v.is_empty() { + cfg.todo_path = v.to_string(); + } + } + // 37 监控轮询:推送目标 chat_id,空 = 只落日志 + if let Some(v) = get(KV_MONITOR_CHAT) { + let v = v.trim().trim_matches('"').to_string(); + if !v.is_empty() { + cfg.monitor_chat = Some(v); + } + } + cfg + } +} + +async fn read_config(app: &AppHandle) -> EngineConfig { + let state = app.state::(); + let kv = [ + (KV_ENABLED, state.settings.get(KV_ENABLED).await.ok().flatten()), + (KV_POLL_SECS, state.settings.get(KV_POLL_SECS).await.ok().flatten()), + (KV_CLAIM_SCOPE, state.settings.get(KV_CLAIM_SCOPE).await.ok().flatten()), + (KV_TODO_PATH, state.settings.get(KV_TODO_PATH).await.ok().flatten()), + (KV_MONITOR_CHAT, state.settings.get(KV_MONITOR_CHAT).await.ok().flatten()), + ]; + EngineConfig::from_kv(&|k| { + kv.iter().find(|(key, _)| *key == k).and_then(|(_, v)| v.clone()) + }) +} + +// ============================================================ +// 看板解析(纯函数) +// ============================================================ + +/// 条目状态标记(`[ ]` 待办 / `[→]` 在途 / `[x]` 完成 / `[!]` 受阻) +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BoardStatus { + Open, + InProgress, + Done, + Blocked, +} + +/// 看板编号条目 +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BoardEntry { + pub number: u64, + pub status: BoardStatus, + /// 正文以 🔴 开头(G2 门禁判据;只认标记后首字符,正文提及「🔴」字样的条目不误判) + pub is_red: bool, + pub text: String, +} + +/// 解析单行编号条目:`N. [标记] 正文`(行首允许缩进;`- [ ]` 破折号清单不在引擎扫描面) +pub fn parse_entry_line(line: &str) -> Option { + let t = line.trim_start(); + let digits_len = t.chars().take_while(|c| c.is_ascii_digit()).count(); + if digits_len == 0 { + return None; + } + let number: u64 = t[..digits_len].parse().ok()?; + let rest = t[digits_len..].strip_prefix(". ")?; + let rest = rest.strip_prefix('[')?; + let marker = rest.chars().next()?; + let status = match marker { + ' ' => BoardStatus::Open, + '→' => BoardStatus::InProgress, + 'x' => BoardStatus::Done, + '!' => BoardStatus::Blocked, + _ => return None, + }; + let rest = &rest[marker.len_utf8()..]; + let rest = rest.strip_prefix(']')?; + // 标记后要求一个空格再接正文(与看板条目格式一致);无空格的紧贴形态按正文兼容 + let text = match rest.strip_prefix(' ') { + Some(text) => text.trim_end().to_string(), + None => rest.trim_end().to_string(), + }; + let is_red = text.starts_with("🔴"); + Some(BoardEntry { + number, + status, + is_red, + text, + }) +} + +/// 解析整板,保持原文行序 +pub fn parse_board(content: &str) -> Vec { + content.lines().filter_map(parse_entry_line).collect() +} + +// ============================================================ +// 领单判据与选择器(纯函数,G2 门禁代码复刻) +// ============================================================ + +/// 领单选择(G2 门禁复刻,不依赖会话自觉): +/// - 存在未清零 🔴(非 Done 的 🔴 条目)→ 只领首个 🔴 待办条目;🔴 在途/受阻无人可领则不领非 🔴(fail-closed) +/// - 🔴 已全部清零 → 按范围领:RedOnly 不领(None),AllOpen 领首个 `[ ]` 条目 +/// - `[→]` 即引擎锁,一律跳过;同轮只返回 1 条(单会话单在途) +pub fn select_claim<'a>(entries: &'a [BoardEntry], scope: ClaimScope) -> Option<&'a BoardEntry> { + let red_uncleared = entries.iter().any(|e| e.is_red && e.status != BoardStatus::Done); + if red_uncleared { + return entries + .iter() + .find(|e| e.is_red && e.status == BoardStatus::Open); + } + match scope { + ClaimScope::RedOnly => None, + ClaimScope::AllOpen => entries.iter().find(|e| e.status == BoardStatus::Open), + } +} + +/// 领单状态改写(纯函数):目标条目 `[ ]` → `[→](自动续作)`,其余行逐字不动(不新增编号)。 +/// 条目不存在或非待办态(引擎锁已被占/已完成/受阻)返回 None 拒改。 +pub fn apply_claim(content: &str, number: u64) -> Option { + let mut changed = false; + let mut out = String::with_capacity(content.len() + 32); + for line in content.lines() { + if !changed { + if let Some(e) = parse_entry_line(line) { + if e.number == number { + if e.status != BoardStatus::Open { + return None; + } + out.push_str(&format!("{}. [→](自动续作) {}", e.number, e.text)); + out.push('\n'); + changed = true; + continue; + } + } + } + out.push_str(line); + out.push('\n'); + } + if !changed { + return None; + } + // lines() 丢弃结尾换行,原文无结尾换行时去补写的最后一个 + if !content.ends_with('\n') { + out.pop(); + } + Some(out) +} + +/// 原子写盘:临时文件(同目录同盘保证 rename 原子) + rename 覆盖,防半写 +fn atomic_write(path: &Path, content: &str) -> std::io::Result<()> { + let file_name = path + .file_name() + .map(|n| n.to_string_lossy().to_string()) + .unwrap_or_else(|| "todo.md".to_string()); + let tmp = path.with_file_name(format!(".{}.b5-tmp", file_name)); + std::fs::write(&tmp, content)?; + // Windows 下 std::fs::rename 走 MOVEFILE_REPLACE_EXISTING,可覆盖已存在目标 + std::fs::rename(&tmp, path) +} + +// ============================================================ +// 引擎共享状态 + 收敛回读钩子 +// ============================================================ + +struct EngineShared { + /// 收敛唤醒:轮询睡眠与完成通知二选一先到,完成即提前扫板(不强实时) + notify: tokio::sync::Notify, + /// 引擎在途会话(单会话单在途的唯一真相源) + in_flight: Mutex>, + /// 连续 spawn 失败计数(成功领单清零) + failures: AtomicU32, + /// 熔断闸:置起后引擎自停(开关保持也不再 spawn),重置=开关关→开 + breaker_tripped: AtomicBool, +} + +static ENGINE: OnceLock = OnceLock::new(); + +fn engine() -> &'static EngineShared { + ENGINE.get_or_init(|| EngineShared { + notify: tokio::sync::Notify::new(), + in_flight: Mutex::new(None), + failures: AtomicU32::new(0), + breaker_tripped: AtomicBool::new(false), + }) +} + +/// 收敛回读钩子:agentic 收敛出口(emit_ai_completed_once,全部正常出口统一单点)调用。 +/// 仅当完成会话是引擎在途会话时清标志并唤醒轮询器;用户自己的会话完成到达时零动作。 +pub fn notify_batch_done(conv_id: &str) { + let e = engine(); + let is_ours = { + let mut guard = e + .in_flight + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let matches = guard.as_deref() == Some(conv_id); + if matches { + *guard = None; + } + matches + }; + if is_ours { + tracing::info!("[B5] 在途会话 {} 收敛,唤醒轮询器续作", conv_id); + e.notify.notify_one(); + } +} + +// ============================================================ +// 轮询调度器 +// ============================================================ + +/// 引擎挂载入口(lib.rs setup 调用一次)。开关关时每轮只读配置空转,零副作用。 +pub fn start(app: AppHandle) { + tauri::async_runtime::spawn(async move { + let cfg = read_config(&app).await; + let mut interval = tokio::time::interval(Duration::from_secs(cfg.poll_secs)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut cur_poll_secs = cfg.poll_secs; + loop { + // 首轮 interval 立即触发(启动即恢复 parked);之后睡眠与收敛通知先到先行 + tokio::select! { + _ = interval.tick() => {} + _ = engine().notify.notified() => {} + } + let cfg = read_config(&app).await; + if cfg.poll_secs != cur_poll_secs { + cur_poll_secs = cfg.poll_secs; + interval = tokio::time::interval(Duration::from_secs(cur_poll_secs)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + } + if !cfg.enabled { + // 开关拉下即复位熔断(重置路径:关→开) + engine().breaker_tripped.store(false, Ordering::SeqCst); + engine().failures.store(0, Ordering::SeqCst); + continue; + } + run_tick(&app, &cfg).await; + } + }); +} + +/// 单轮:熔断检查 → 在途检查 → 冻结标志同步 → parked 恢复优先 → 看板领单 +async fn run_tick(app: &AppHandle, cfg: &EngineConfig) { + let e = engine(); + if e.breaker_tripped.load(Ordering::SeqCst) { + return; // 熔断自停,等人工开关复位 + } + // 提交冻结标志同步(39 治理):每轮按看板 🔒 行 + KV 覆盖同步到执行闸, + // 引擎派发会话内任何 git commit 在 spawn 前被拒(改动留工作区)。 + sync_commit_freeze(app, cfg).await; + // 在途判据(双保险):引擎标志为真相,can_accept_request 兜底判残留—— + // 异常退出漏走收敛钩子时会话实际已空闲,清残留继续,引擎不因泄漏永久卡死。 + let in_flight = e + .in_flight + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + if let Some(conv) = in_flight { + let state = app.state::(); + if state.conv_states.can_accept_request(&conv) { + tracing::warn!("[B5] 在途标志残留(会话 {} 已非生成态),自动清除", conv); + *e.in_flight + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + } else { + return; // 真在途:本轮跳过 + } + } + // parked 自动恢复优先于领单(断点续跑先于新任务) + match recover_one_parked(app).await { + Ok(conv) => { + set_in_flight(conv); + return; + } + Err(err) => { + // "无可恢复"是常态非失败;其余恢复失败计熔断(spawn 同链路) + if err != RECOVER_NONE { + tracing::warn!("[B5] parked 恢复失败: {}", err); + count_spawn_failure(); + return; + } + } + } + // 领单 + if let Err(err) = claim_and_spawn(app, cfg).await { + // 熔断已在计数处置时打 error,这里只记过程 + tracing::warn!("[B5] 本轮领单未成: {}", err); + } + // 37 监控轮询:每轮末尾自监控快照(活性/熔断/失败计数),异常推飞书,正常落日志 + run_self_monitor(app, cfg).await; + // D2d 归档清理:每日一次(挂钟桶去重),清 90 天前外置派生文件(不碰 DB 行) + run_daily_offload_cleanup(app).await; +} + +/// D2d 每日归档清理(挂钟桶去重:同一天只跑一次;失败静默,下一天重试)。 +async fn run_daily_offload_cleanup(app: &AppHandle) { + use std::sync::atomic::AtomicU64; + static LAST_DAY: AtomicU64 = AtomicU64::new(0); + let day = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() / 86400) + .unwrap_or(0); + if LAST_DAY.swap(day, Ordering::SeqCst) == day { + return; // 今日已跑 + } + let data_dir = app.path().app_data_dir().unwrap_or_default(); + let (removed, freed) = crate::commands::ai::tool_result_offload::cleanup_expired(&data_dir); + if removed > 0 { + tracing::info!("[D2d] 归档清理:删除 {removed} 个过期外置文件,释放 {freed} 字节"); + } +} + +// ============================================================ +// 37 自监控:周期快照 + 异常通报(todo 37 验收「1 个周期内被发现并通报」) +// ============================================================ + +/// 同段异常只告警一次(恢复后复位);静默重复告警=刷屏,非静默漏报=盲区 +static MONITOR_ALERTED: OnceLock = OnceLock::new(); +fn monitor_alerted() -> &'static AtomicBool { + MONITOR_ALERTED.get_or_init(|| AtomicBool::new(false)) +} + +/// 单轮自监控快照:熔断在场/连续失败≥2(未熔断预警)/在途会话状态,汇总一行。 +/// 异常:推 monitor_chat(飞书,若配)+error 日志,同段异常只推一次;正常:tracing 一行。 +/// 本进程消失=监控同死,由外部观察者(飞书侧静默)兜底——通报出口设计为飞书可见, +/// 引擎死了飞书自然静默,用户侧「该来没来」即是信号(与 u-claw 进度卡治理同哲学)。 +async fn run_self_monitor(app: &AppHandle, cfg: &EngineConfig) { + let e = engine(); + let failures = e.failures.load(Ordering::SeqCst); + let tripped = e.breaker_tripped.load(Ordering::SeqCst); + let in_flight = e + .in_flight + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + let healthy = !tripped && failures < 2; + let snapshot = match (&in_flight, tripped, failures) { + (_, true, n) => format!("🔴 熔断中(累计失败 {n}),引擎自停——需人工关开 df-unattended-enabled 复位"), + (_, false, n) if n >= 2 => format!("🟡 连续 spawn 失败 {n} 次(阈值 3),即将熔断"), + (Some(conv), _, _) => format!("🟢 引擎健康,在途会话 {conv}"), + (None, _, _) => format!("🟢 引擎健康,空闲"), + }; + if healthy { + tracing::info!("[B5][监控] {snapshot}"); + monitor_alerted().store(false, Ordering::SeqCst); + return; + } + // 异常段:同段只推一次(恢复即复位) + if monitor_alerted().swap(true, Ordering::SeqCst) { + return; + } + tracing::error!("[B5][监控] {snapshot}"); + crate::commands::ai::feishu_notify::push_monitor_alert( + app, + cfg.monitor_chat.as_deref().unwrap_or(""), + &snapshot, + ) + .await; +} + +const RECOVER_NONE: &str = "无可恢复 parked 会话"; + +/// 提交冻结标志同步(39 治理):读 KV 覆盖标志 + 看板 🔒 行,写入执行闸静态位。 +/// 引擎派发的会话内 git commit 在 df-execute spawn 前被拒;用户本人终端不受影响。 +async fn sync_commit_freeze(app: &AppHandle, cfg: &EngineConfig) { + let state = app.state::(); + let kv_frozen = state.settings.get(KV_COMMIT_FROZEN).await.ok().flatten(); + let board = std::fs::read_to_string(&cfg.todo_path).ok(); + let frozen = resolve_frozen(board.as_deref(), &|_| kv_frozen.clone()); + df_execute::commit_freeze::set_frozen(frozen); + if frozen { + tracing::info!("[B5] 提交冻结令在场(看板 🔒 或 KV df-commit-frozen),引擎会话 git commit 已闸断"); + } +} + +fn set_in_flight(conv_id: String) { + *engine() + .in_flight + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(conv_id); +} + +/// spawn 失败计数:连续 3 次熔断(引擎自停 + error 告警,开关保持) +fn count_spawn_failure() { + let e = engine(); + let n = e.failures.fetch_add(1, Ordering::SeqCst) + 1; + if n >= MAX_CONSECUTIVE_FAILURES && !e.breaker_tripped.swap(true, Ordering::SeqCst) { + tracing::error!( + "[B5] 连续 {} 次 spawn 失败,引擎熔断自停(开关保持但不再领单/恢复);人工复位=将 df-unattended-enabled 关后重开", + n + ); + } +} + +/// parked 会话自动恢复:守卫链与 ai_recover_parked(commands/conversation.rs,IPC 正典) +/// 同款(parked 态校验 + can_accept_request 双保险),差异仅两点——引擎批量取 +/// list_parked 首个(最近更新优先)、pin_active=false 不劫持用户活跃面板。 +/// 恢复引导消息复用正典常量,文案单一来源。 +async fn recover_one_parked(app: &AppHandle) -> Result { + let state = app.state::(); + let parked = state + .ai_conversations + .list_parked() + .await + .map_err(|e| e.to_string())?; + let Some(record) = parked.into_iter().next() else { + return Err(RECOVER_NONE.to_string()); + }; + let conv_id = record.id; + if record.loop_state.as_deref() != Some("parked") { + return Err(format!("会话 {} 非可恢复状态", conv_id)); + } + if !state.conv_states.can_accept_request(&conv_id) { + return Err(format!("会话 {} 正在生成中,无需恢复", conv_id)); + } + spawn_agentic_session( + app, + &state, + crate::commands::ai::commands::conversation::PARKED_RECOVERY_MESSAGE.to_string(), + None, + None, + None, + Some(conv_id.clone()), + None, + None, + false, + ) + .await + .map(|_| conv_id) +} + +/// 看板领单:读板 → 选条目 → 原子写领单标记 → 正典链路开工;spawn 失败回滚领单标记 +async fn claim_and_spawn(app: &AppHandle, cfg: &EngineConfig) -> Result<(), String> { + let path = Path::new(&cfg.todo_path); + let content = std::fs::read_to_string(path) + .map_err(|e| format!("看板读取失败 {}: {}", cfg.todo_path, e))?; + let entries = parse_board(&content); + let Some(entry) = select_claim(&entries, cfg.claim_scope) else { + return Err("无可领条目".to_string()); + }; + let (number, text) = (entry.number, entry.text.clone()); + let new_content = + apply_claim(&content, number).ok_or_else(|| format!("条目 #{} 不可领", number))?; + atomic_write(path, &new_content).map_err(|e| format!("看板领单写入失败: {}", e))?; + tracing::info!("[B5] 已领看板条目 #{}: {}", number, truncate(&text, 80)); + // 领单任务书:条目原文全文 + 完成指令(改 [x] 附验收输出) + 冻结期硬约束(39 治理) + let freeze_clause = if df_execute::commit_freeze::is_frozen(None) { + "\n\n⚠️ 【提交冻结令(硬约束)】当前处于提交冻结期(看板 🔒 令),禁止执行任何 git commit/\ + git push——所有改动一律保留在工作区,绝不提交;解冻须等用户「攒批提交通知」。\ + 执行层已设机械闸:冻结在场时 commit 命令会被直接拒绝。\n" + } else { + "" + }; + let message = format!( + "[无人值守自动领单] 看板条目 #{}:\n\n{}\n\n要求:\n\ + 1. 按条目内容完成任务(看板文件: {},仓库即该文件所在仓)。\n\ + 2. 完成后将该条目状态标记改为 [x](保持编号与正文不变,仅改状态标记),并附验收输出(验证命令+实测结果)。\n\ + 3. 无法完成或需人工拍板时,保持条目标记 [→] 并在输出中说明原因,等待人工处理。{}", + number, text, cfg.todo_path, freeze_clause + ); + let state = app.state::(); + match spawn_agentic_session( + app, + &state, + message, + None, // language 沿用默认 + None, // skill 不挂 + None, // model_override 沿用路由 + Some(df_types::types::new_id()), // 引擎独立后台会话,不共用用户活跃面板 + None, + None, + false, // pin_active=false 不劫持活跃面板 + ) + .await + { + Ok(conv_id) => { + engine().failures.store(0, Ordering::SeqCst); + set_in_flight(conv_id); + Ok(()) + } + Err(spawn_err) => { + // 领单标记回滚(spawn 未成,板不得留假在途锁):best-effort 重写原文 + if let Err(e) = atomic_write(path, &content) { + tracing::warn!("[B5] 条目 #{} 领单标记回滚失败(需人工核对): {}", number, e); + } + count_spawn_failure(); + Err(format!("条目 #{} spawn 失败: {}", number, spawn_err)) + } + } +} + +fn truncate(s: &str, max: usize) -> String { + if s.chars().count() <= max { + s.to_string() + } else { + let cut: String = s.chars().take(max).collect(); + format!("{}…", cut) + } +} + +// ============================================================ +// 单测(纯函数:解析/🔴 判据/门禁选择/领单改写) +// ============================================================ +#[cfg(test)] +mod tests { + use super::*; + + fn entry(n: u64, status: BoardStatus, red: bool, text: &str) -> BoardEntry { + BoardEntry { + number: n, + status, + is_red: red, + text: text.to_string(), + } + } + + #[test] + fn parse_statuses_and_red_marker() { + let board = "前言不解析\n1. [ ] 普通待办\n2. [→] 在途条目(自动续作)\n\ + 3. [x] 已完成 🔴 清零\n4. [!] 受阻条目 🔴\n5. [x] 🔴 完成红\n\ + 6. [ ] 描述里提到 🔴 字样的条目不误判\n\ + - [ ] 破折号清单不在扫描面\n2026-06-18 归档行不解析\n"; + let entries = parse_board(board); + assert_eq!( + entries, + vec![ + entry(1, BoardStatus::Open, false, "普通待办"), + entry(2, BoardStatus::InProgress, false, "在途条目(自动续作)"), + entry(3, BoardStatus::Done, false, "已完成 🔴 清零"), + entry(4, BoardStatus::Blocked, false, "受阻条目 🔴"), + entry(5, BoardStatus::Done, true, "🔴 完成红"), + entry(6, BoardStatus::Open, false, "描述里提到 🔴 字样的条目不误判"), + ] + ); + } + + #[test] + fn parse_requires_numbered_entry_shape() { + assert!(parse_entry_line("1.[ ] 无空格点号不匹配").is_none()); + assert!(parse_entry_line("12. [y] 未知标记不匹配").is_none()); + assert!(parse_entry_line("abc. [ ] 非数字编号").is_none()); + // 标记后无空格的紧贴形态按正文兼容不丢条目 + assert_eq!( + parse_entry_line("7. [x]**紧贴**").map(|e| e.text), + Some("**紧贴**".to_string()) + ); + } + + #[test] + fn g2_gate_red_blocks_non_red_claims() { + // 🔴 未清零:即便更早存在普通待办,也只能领 🔴 + let board = vec![ + entry(1, BoardStatus::Open, false, "普通"), + entry(2, BoardStatus::Open, true, "红待办"), + entry(3, BoardStatus::Done, true, "红完成"), + ]; + assert_eq!(select_claim(&board, ClaimScope::AllOpen).map(|e| e.number), Some(2)); + } + + #[test] + fn g2_gate_red_in_flight_fails_closed() { + // 🔴 在场但无待办红(在途/受阻)→ 不领非 🔴,fail-closed + let board = vec![ + entry(1, BoardStatus::Open, false, "普通"), + entry(2, BoardStatus::InProgress, true, "红在途"), + entry(3, BoardStatus::Blocked, true, "红受阻"), + ]; + assert_eq!(select_claim(&board, ClaimScope::AllOpen), None); + assert_eq!(select_claim(&board, ClaimScope::RedOnly), None); + } + + #[test] + fn g2_gate_red_cleared_then_scope_decides() { + // 🔴 全清零:red_only 不领,all_open 领首条待办 + let board = vec![ + entry(1, BoardStatus::Done, true, "红完成"), + entry(2, BoardStatus::InProgress, false, "在途跳过"), + entry(3, BoardStatus::Open, false, "可领"), + ]; + assert_eq!(select_claim(&board, ClaimScope::RedOnly), None); + assert_eq!(select_claim(&board, ClaimScope::AllOpen).map(|e| e.number), Some(3)); + } + + #[test] + fn g2_gate_engine_lock_skipped() { + // [→] 即引擎锁:open 红优先于更靠前的在途红,在途条目一律不可再领 + let board = vec![ + entry(1, BoardStatus::InProgress, true, "红在途"), + entry(2, BoardStatus::Open, true, "红可领"), + ]; + assert_eq!(select_claim(&board, ClaimScope::RedOnly).map(|e| e.number), Some(2)); + assert_eq!(select_claim(&[], ClaimScope::AllOpen), None); + } + + #[test] + fn claim_rewrites_only_target_marker() { + let board = "1. [ ] 甲\n2. [ ] 乙 🔴\n3. [x] 丙\n尾部普通行\n"; + let claimed = apply_claim(board, 2).expect("可领"); + assert_eq!( + claimed, + "1. [ ] 甲\n2. [→](自动续作) 乙 🔴\n3. [x] 丙\n尾部普通行\n" + ); + // 无结尾换行原文保持无结尾换行 + assert_eq!( + apply_claim("1. [ ] 甲", 1).as_deref(), + Some("1. [→](自动续作) 甲") + ); + } + + #[test] + fn claim_rejects_non_open_or_missing() { + let board = "1. [→] 在途\n2. [x] 完成\n3. [!] 受阻\n"; + assert_eq!(apply_claim(board, 1), None, "在途(引擎锁)不可再领"); + assert_eq!(apply_claim(board, 2), None, "完成不可领"); + assert_eq!(apply_claim(board, 3), None, "受阻不可领"); + assert_eq!(apply_claim(board, 9), None, "不存在不可领"); + } + + #[test] + fn config_parses_kv_with_defaults_and_clamps() { + let cfg = EngineConfig::from_kv(&|_| None); + assert_eq!(cfg, EngineConfig::default()); + let cfg = EngineConfig::from_kv(&|k| match k { + "df-unattended-enabled" => Some("\"true\"".to_string()), + "df-unattended-poll-secs" => Some("5".to_string()), + "df-unattended-claim-scope" => Some("all_open".to_string()), + "df-unattended-todo-path" => Some("D:\\x\\todo.md".to_string()), + _ => None, + }); + assert!(cfg.enabled); + assert_eq!(cfg.poll_secs, 30, "低于下限钳到 30s 防热循环"); + assert_eq!(cfg.claim_scope, ClaimScope::AllOpen); + assert_eq!(cfg.todo_path, "D:\\x\\todo.md"); + } + + /// 39 冻结判据:看板 🔒 行生效/无 🔒 不冻结/KV 显式覆盖双向 + #[test] + fn commit_freeze_board_and_kv_judge() { + let frozen_board = "# DevFlow 工作看板\n> 🔒 **提交冻结**:期间只改不提\n1. [ ] 甲\n"; + let open_board = "# 看板\n1. [ ] 甲\n> 冻结(文字提及但非 🔒 行首)\n"; + assert!(board_frozen(frozen_board)); + assert!(!board_frozen(open_board), "非 🔒 行首不误判"); + // KV 显式覆盖双向 + assert!(resolve_frozen(Some(open_board), &|k| { + (k == "df-commit-frozen").then(|| "true".to_string()) + })); + assert!(!resolve_frozen(Some(frozen_board), &|k| { + (k == "df-commit-frozen").then(|| "false".to_string()) + }), "KV 显式解冻覆盖看板 🔒 行"); + // 无 KV 时按看板 + assert!(resolve_frozen(Some(frozen_board), &|_| None)); + assert!(!resolve_frozen(Some(open_board), &|_| None)); + assert!(!resolve_frozen(None, &|_| None), "无板无 KV 默认不冻结"); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 21ba939..676e333 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -68,10 +68,16 @@ pub fn run() { // spawn_agent 子代理循环需要 AppHandle(emit/guard),工具注册期无法捕获, // setup 阶段回注进程级 OnceLock(重复注入忽略,首次生效)。 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()); // 飞书直连渠道:配置启用时起长连接(读 feishu-config;未配置/disabled 静默跳过) feishu_channel::start_if_enabled(app.handle().clone()); + // B5 无人值守续作引擎:看板轮询自动领单 + parked 自动恢复。 + // 常驻轮询循环,开关(df-unattended-enabled,默认关)关时每轮只读配置空转零副作用。 + commands::ai::unattended::start(app.handle().clone()); + // ── Phase3: 内嵌 HTTP MCP server + 托盘常驻 + 单实例 ── // 1) MCP 运行时状态:先 manage 初始 running=false,spawn 绑定成功后更新为 true。 // mcp_get_status IPC 经 State> 读取,spawn 内写锁同步。