新增: 上下文管理演进 Wave1(契约定义 + T1动态阈值 + T2命名空间)

Agent 3: 共享契约类型定义
- CompressedSummary(T3输出格式)
- WorkflowContextBlock(T4工作流DAG上下文)
- WorkingContext + GoalItem/DecisionItem/StepInfo(T5常驻上下文)
- 脏标记版本追踪 + reset_all 压缩联动

T1: 动态压缩阈值
- maybe_auto_compress 加 sys_tokens 参数,触发条件从 budget*0.6 改为 available*0.6
- 调用处传 sys_tokens(loop顶部已缓存)
- 3 行核心改动

T2: 工具命名空间存储
- crates/df-ai/src/namespace_store.rs: 完整实现 + 单元测试
- process_tool_calls: Low/Med/High 路径 namespace 路由
- save_conversation: namespace 引用展开为原文再落 DB
- AiSession 新增 namespace_store 字段
This commit is contained in:
lxy
2026-07-20 01:05:20 +08:00
parent b1d7deece1
commit faa61a9ba5
10 changed files with 805 additions and 6 deletions
+167 -1
View File
@@ -1,4 +1,4 @@
//! 上下文管理纯函数与数据类型 — 从 context.rs 抽离的无 `self` 依赖部分 //! 上下文管理纯函数与数据类型 — 从 context.rs 抽离的无 `self` 依赖部分
//! //!
//! 职责: //! 职责:
//! - Token 粗估器(字符级近似,无 tokenizer 依赖) //! - Token 粗估器(字符级近似,无 tokenizer 依赖)
@@ -553,6 +553,9 @@ pub const TOOL_MISSING_PREFIX: &str = "tool_missing_";
/// 兜底:sanitize view-only 不改持久化,失败只影响单请求,flag 关→行为完全等价改动前。 /// 兜底:sanitize view-only 不改持久化,失败只影响单请求,flag 关→行为完全等价改动前。
pub const PLACEHOLDER_INTEGRITY_ENABLED: bool = true; pub const PLACEHOLDER_INTEGRITY_ENABLED: bool = true;
/// namespace 引用路径前缀(转发至 namespace_store,供 context.rs 统一路径引用)
pub const NAMESPACE_REF_PREFIX: &str = crate::namespace_store::NAMESPACE_REF_PREFIX;
/// 占位 tool_result 内容中嵌入的唯一标记前缀(供 sanitize 识别"审批挂起占位,不可裁")。 /// 占位 tool_result 内容中嵌入的唯一标记前缀(供 sanitize 识别"审批挂起占位,不可裁")。
/// ///
/// 写入处(audit/cache.rs:PENDING_APPROVAL_PLACEHOLDER)格式:`{占位文本}{__PENDING__}{tc_id}`, /// 写入处(audit/cache.rs:PENDING_APPROVAL_PLACEHOLDER)格式:`{占位文本}{__PENDING__}{tc_id}`,
@@ -604,6 +607,169 @@ pub fn extract_pending_tc_id(content: &str) -> Option<&str> {
.filter(|id| !id.is_empty()) .filter(|id| !id.is_empty())
} }
// ============================================================
// Wave 共享契约类型(由 上下文管理演进-执行计划 Agent 3 定义)
// ============================================================
/// 压缩摘要双格式(T3 输出, T5 消费)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CompressedSummary {
/// JSON 卡片序列化(主题/决策/待办/token节省)
pub json_card: String,
/// 自然语言摘要(向前兼容关键词兜底降级)
pub nl_summary: String,
}
/// 工作流 DAG 上下文块(T4 输出 → 注入 system prompt
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowContextBlock {
pub workflow_id: String,
pub workflow_name: String,
pub total_nodes: usize,
pub completed_nodes: usize,
pub current_node: Option<String>,
pub next_nodes: Vec<String>,
}
impl WorkflowContextBlock {
/// 格式化为 system prompt 可读文本
pub fn to_system_text(&self) -> String {
let mut s = format!("工作流: {} (ID: {})", self.workflow_name, self.workflow_id);
s.push_str(&format!("\n进度: {}/{}", self.completed_nodes, self.total_nodes));
if let Some(ref cur) = self.current_node {
s.push_str(&format!("\n当前步骤: {}", cur));
}
if !self.next_nodes.is_empty() {
s.push_str(&format!("\n后续: {}", self.next_nodes.join("")));
}
s
}
}
/// 目标状态(T5 WorkingContext
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum GoalStatus {
Active,
Completed,
Cancelled,
}
/// 步骤状态
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum StepStatus {
Pending,
InProgress,
Done,
Blocked,
}
/// 目标条目
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GoalItem {
pub id: String,
pub text: String,
pub status: GoalStatus,
}
/// 决策条目
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DecisionItem {
pub question: String,
pub decision: String,
pub reason: String,
}
/// 步骤信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StepInfo {
pub step_name: String,
pub status: StepStatus,
}
/// WorkingContext 版本号(脏标记追踪)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkingContextVersion {
pub goals_changed_at: u32,
pub decisions_changed_at: u32,
pub unresolved_changed_at: u32,
pub step_changed_at: u32,
pub current_turn: u32,
}
impl WorkingContextVersion {
pub fn new() -> Self {
Self {
goals_changed_at: 0,
decisions_changed_at: 0,
unresolved_changed_at: 0,
step_changed_at: 0,
current_turn: 0,
}
}
/// 压缩后重置:将所有字段标记为本轮变更
pub fn reset_all(&mut self, _summary: &str, current_turn: u32) {
self.goals_changed_at = current_turn;
self.decisions_changed_at = current_turn;
self.unresolved_changed_at = current_turn;
self.current_turn = current_turn;
}
}
/// WorkingContext 常驻上下文(T5 实现)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkingContext {
pub goals: Vec<GoalItem>,
pub decisions: Vec<DecisionItem>,
pub unresolved: Vec<String>,
pub current_step: Option<StepInfo>,
pub version: WorkingContextVersion,
}
impl WorkingContext {
pub fn new() -> Self {
Self {
goals: Vec::new(),
decisions: Vec::new(),
unresolved: Vec::new(),
current_step: None,
version: WorkingContextVersion::new(),
}
}
/// 构建本轮需要注入的 L1 内容(L1a 必带 + L1b 条件注入)
pub fn build_injection(&self, recent_turns: u32) -> String {
let mut parts: Vec<String> = Vec::new();
// L1a: 当前步骤(每轮必带)
if let Some(ref step) = self.current_step {
parts.push(format!("当前步骤: {} [{:?}]", step.step_name, step.status));
}
// L1b: 过去 N 轮内有变更才注入
let threshold = self.version.current_turn.saturating_sub(recent_turns);
let has_active = self.goals.iter().any(|g| g.status == GoalStatus::Active);
if has_active && self.version.goals_changed_at >= threshold {
let active: Vec<&str> = self.goals.iter()
.filter(|g| g.status == GoalStatus::Active)
.map(|g| g.text.as_str())
.collect();
parts.push(format!("目标({}): {}", active.len(), active.join("; ")));
}
if self.version.decisions_changed_at >= threshold && !self.decisions.is_empty() {
let recent: Vec<&str> = self.decisions.iter().map(|d| d.decision.as_str()).collect();
parts.push(format!("近期决策: {}", recent.join("; ")));
}
if self.version.unresolved_changed_at >= threshold && !self.unresolved.is_empty() {
parts.push(format!("待解: {}", self.unresolved.join("; ")));
}
parts.join("\n")
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
+1
View File
@@ -42,6 +42,7 @@ pub mod router;
// 避免重写退避/分类逻辑(对齐决策 F-260616-07 a1)。改 pub mod 后对外仅暴露纯函数 + 常量。 // 避免重写退避/分类逻辑(对齐决策 F-260616-07 a1)。改 pub mod 后对外仅暴露纯函数 + 常量。
pub mod retry; pub mod retry;
pub mod sse_parser; pub mod sse_parser;
pub mod namespace_store;
use provider::LlmProvider; use provider::LlmProvider;
use reqwest::Client; use reqwest::Client;
+251
View File
@@ -0,0 +1,251 @@
//! 工具命名空间存储 — 大工具结果不进主消息队列
//!
//! 当工具执行结果超过阈值时,存入 `NamespaceStore`,主队列只保留轻量引用路径。
//! 运行时为内存 HashMap(LRU 淘汰),持久化时由调用方展开引用写 DB。
//!
//! 存储格式: "namespace://tool_name/args_hash"
//! 例: namespace://read_file/a1b2c3d4
use std::collections::HashMap;
use std::time::Instant;
/// 引用路径前缀
pub const NAMESPACE_REF_PREFIX: &str = "namespace://";
/// namespace 字节阈值(> 2048 bytes 触发)
pub const NAMESPACE_BYTE_THRESHOLD: usize = 2048;
/// namespace 行数阈值(> 50 行 触发)
pub const NAMESPACE_LINE_THRESHOLD: usize = 50;
/// 始终进 namespace 的工具名(其输出几乎总是大结果)
pub const ALWAYS_LARGE_TOOLS: &[&str] = &["read_file", "list_directory", "grep", "diff_files"];
/// 判定工具结果是否应进入 namespace
pub fn should_use_namespace(content: &str, tool_name: &str) -> bool {
if ALWAYS_LARGE_TOOLS.contains(&tool_name) {
return true;
}
content.len() > NAMESPACE_BYTE_THRESHOLD
|| content.lines().count() > NAMESPACE_LINE_THRESHOLD
}
/// 判定 tool_result 内容是否为 namespace 引用路径
pub fn is_namespace_ref(content: &str) -> bool {
content.starts_with(NAMESPACE_REF_PREFIX)
}
/// 从引用路径中提取内部 key
pub fn parse_namespace_key(full_path: &str) -> Option<&str> {
full_path.strip_prefix(NAMESPACE_REF_PREFIX)
.and_then(|s| s.split('/').nth(1))
}
/// namespace 条目
#[derive(Debug, Clone)]
pub struct NamespaceEntry {
pub key: String,
pub tool_name: String,
pub content: String,
pub created_at: Instant,
pub access_count: u64,
}
/// namespace 存储(内存 HashMap + LRU 淘汰)
#[derive(Debug, Clone)]
pub struct NamespaceStore {
entries: HashMap<String, NamespaceEntry>,
max_bytes: usize,
current_bytes: usize,
}
impl NamespaceStore {
/// 创建 namespace 存储
pub fn new(max_bytes: usize) -> Self {
Self {
entries: HashMap::new(),
max_bytes,
current_bytes: 0,
}
}
/// 默认大小:10MB
pub fn default() -> Self {
Self::new(10 * 1024 * 1024)
}
/// 存入 namespace。返回引用路径字符串。
pub fn store(&mut self, tool_name: &str, content: &str) -> String {
let key = simple_hash(content);
let path = format!("{}{}/{}", NAMESPACE_REF_PREFIX, tool_name, key);
// 已存在 → 刷新 access_count,不重复占用空间
if self.entries.contains_key(&key) {
if let Some(entry) = self.entries.get_mut(&key) {
entry.access_count += 1;
}
return path;
}
// 检查容量,超限淘汰
let new_bytes = content.len();
self.evict_if_needed(new_bytes);
let entry = NamespaceEntry {
key: key.clone(),
tool_name: tool_name.to_string(),
content: content.to_string(),
created_at: Instant::now(),
access_count: 1,
};
self.current_bytes += new_bytes;
self.entries.insert(key, entry);
path
}
/// 读取完整内容
pub fn read(&mut self, full_path: &str) -> Option<&str> {
let key = parse_namespace_key(full_path)?;
let entry = self.entries.get_mut(key)?;
entry.access_count += 1;
Some(entry.content.as_str())
}
/// 读取(只读,不改 access_count
pub fn read_only(&self, full_path: &str) -> Option<&str> {
let key = parse_namespace_key(full_path)?;
self.entries.get(key).map(|e| e.content.as_str())
}
/// 检查 key 是否存在
pub fn contains(&self, full_path: &str) -> bool {
parse_namespace_key(full_path)
.and_then(|k| self.entries.get(k))
.is_some()
}
/// 条目数
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
/// LRU 淘汰:删除最旧条目直到腾出所需空间
fn evict_if_needed(&mut self, needed: usize) {
while self.current_bytes + needed > self.max_bytes && !self.entries.is_empty() {
// 找最旧的条目
let oldest_key = self.entries.iter()
.min_by_key(|(_, e)| e.created_at)
.map(|(k, _)| k.clone());
if let Some(key) = oldest_key {
if let Some(entry) = self.entries.remove(&key) {
self.current_bytes = self.current_bytes.saturating_sub(entry.content.len());
}
}
}
}
/// 清空
pub fn clear(&mut self) {
self.entries.clear();
self.current_bytes = 0;
}
}
/// 简单的基于长度的哈希(唯一性由 content 长度 + 前 16 字符保证)
/// 用于构建引用路径中的 key 段,不需要密码学强度
fn simple_hash(content: &str) -> String {
let prefix: String = content.chars().take(16).collect();
format!("{}_{}", content.len(), prefix)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn should_use_namespace_small_content() {
assert!(!should_use_namespace("short", "custom_tool"));
}
#[test]
fn should_use_namespace_large_bytes() {
let large = "x".repeat(3000);
assert!(should_use_namespace(&large, "custom_tool"));
}
#[test]
fn should_use_namespace_always_large_tool() {
assert!(should_use_namespace("short", "read_file"));
assert!(should_use_namespace("short", "list_directory"));
}
#[test]
fn should_use_namespace_many_lines() {
let many = (0..60).map(|i| format!("line {}", i)).collect::<Vec<_>>().join("\n");
assert!(should_use_namespace(&many, "custom_tool"));
}
#[test]
fn is_namespace_ref_detects_prefix() {
assert!(is_namespace_ref("namespace://read_file/abc"));
assert!(!is_namespace_ref("file content here"));
}
#[test]
fn parse_namespace_key_extracts_key() {
assert_eq!(parse_namespace_key("namespace://read_file/abc_123"), Some("abc_123"));
assert_eq!(parse_namespace_key("namespace://grep/xyz"), Some("xyz"));
assert_eq!(parse_namespace_key("no_prefix"), None);
}
#[test]
fn store_and_read() {
let mut ns = NamespaceStore::new(100_000);
let path = ns.store("read_file", "hello world content");
assert!(is_namespace_ref(&path));
assert!(path.contains("read_file"));
assert_eq!(ns.read(&path), Some("hello world content"));
}
#[test]
fn store_duplicate_key_does_not_double_count() {
let mut ns = NamespaceStore::new(100_000);
let content = "test content for duplicate check";
let p1 = ns.store("tool_a", content);
let p2 = ns.store("tool_b", content);
assert_eq!(ns.len(), 1, "相同内容应去重");
assert_eq!(ns.read(&p1), ns.read(&p2));
}
#[test]
fn eviction_oldest_removed_when_over_limit() {
let mut ns = NamespaceStore::new(100); // 极小容量
let small = "a"; // 1 byte
ns.store("t1", small);
assert_eq!(ns.len(), 1);
// 填充到超限
let big = "x".repeat(200);
ns.store("t2", &big);
// t1 可能被淘汰
if ns.current_bytes > 100 {
assert!(ns.len() < 2, "超限时应已淘汰旧条目");
}
}
#[test]
fn clear_resets_all() {
let mut ns = NamespaceStore::new(100_000);
ns.store("t1", "content");
assert!(!ns.is_empty());
ns.clear();
assert!(ns.is_empty());
assert_eq!(ns.current_bytes, 0);
}
}
@@ -0,0 +1,352 @@
# 上下文管理演进 — 执行计划
> 关联: `上下文管理演进-任务技术设计-2026-07-20.md` | 状态: 待执行
---
## Wave 划分
3 个 Wave,每个 Wave 内 agent 可并行,Wave 间串行 gate。
```mermaid
graph LR
subgraph "Wave 1 (4 agent 并行)"
A1["Agent 1: T1 动态阈值"]
A2["Agent 2: T2 命名空间存储"]
A3["Agent 3: 接口契约定义"]
end
A3 -.->|"定义 downstream 接口"| A2
A1 --> G1["Gate 1: cargo check + 单测"]
A2 --> G1
G1 --> subgraph "Wave 2 (3 agent 并行)"
B1["Agent 4: T3 结构化摘要"]
B2["Agent 5: T4 工作流 DAG"]
end
B1 --> G2["Gate 2: cargo check + 集成验证"]
B2 --> G2
G2 --> subgraph "Wave 3 (2 agent 串行)"
C1["Agent 6: T5 WorkingContext"]
end
```
---
## Wave 1:基础改造(4 agent 并行)
### Agent 1T1 动态压缩阈值
**改动清单**:精确到行
| 文件 | 行 | 操作 |
|------|----|------|
| `src-tauri/src/commands/ai/agentic/context_lifecycle.rs` | 64 签名 | `maybe_auto_compress``sys_tokens: u32` 参数 |
| 同上 | 88-92 触发条件 | `let budget``let available = budget.saturating_sub(sys_tokens)`;比较用 `available` 替代 `budget` |
| `src-tauri/src/commands/ai/agentic/mod.rs` | ~1117 调用处 | `maybe_auto_compress(...)``sys_tokens` 实参(同作用域 line 964 已定义) |
**接口契约**`maybe_auto_compress` 签名变更——调用方需传入 `sys_tokens`。不影响其他模块。
**验证**`cargo check -p devflow`;观察日志 `available` 值在 system prompt 大时显著小于 `budget`
**回退**:删参数 + 恢复 `budget` 变量,2 行。
---
### Agent 2T2 工具命名空间存储
**接口契约**Wave 1 必须先执行):
```rust
// crates/df-ai/src/namespace_store.rs — 新增,其他 agent 依赖此接口
pub struct NamespaceStore { .. }
impl NamespaceStore {
pub fn new(max_bytes: usize) -> Self;
pub fn store(&mut self, key: &str, tool_name: &str, args: &Value, content: &str) -> String;
pub fn read(&mut self, full_path: &str) -> Option<&str>;
pub fn read_summary(&self, full_path: &str) -> Option<&str>;
}
pub fn should_use_namespace(content: &str, tool_name: &str) -> bool;
pub const NAMESPACE_REF_PREFIX: &str = "namespace://";
pub const NAMESPACE_BYTE_THRESHOLD: usize = 2048;
```
**接线改动**
| 文件 | 行 | 操作 |
|------|----|------|
| `crates/df-ai/src/namespace_store.rs` | 新增 | 完整实现 + 单元测试 |
| `crates/df-ai/src/lib.rs` | — | `pub mod namespace_store;` + `pub use` 重导出 |
| `crates/df-ai/src/context_helpers.rs` | — | 新增 `NAMESPACE_REF_PREFIX``should_use_namespace``is_namespace_ref` |
| `src-tauri/src/commands/ai/audit/mod.rs` | `process_tool_calls` ~380 行 | tool 执行完成后:`if should_use_namespace``ns.store(...)` → push 引用;else → push 原文 |
| `src-tauri/src/commands/ai/mod.rs``state.rs` | `AiSession` struct | 新增 `namespace_store: NamespaceStore` 字段 |
| `src-tauri/src/commands/ai/conversation.rs` | `save_conversation` | namespace 引用展开为原文再写 DB |
**验证**
1. `cargo test -p df-ai``namespace_store` 单元测试(store → read → LRU 淘汰)
2. 手动:`read_file("large.txt")` → DB 中 `tool_result` 内容为原文(引用已展开)
3. `history_tokens` 统计应显著下降(大工具结果只计引用 token)
**与 Agent 3 的边界**`namespace_store.rs` 实现不依赖 AiSession,纯数据结构的 `HashMap` 操作,零耦合,可独立开发和测试。接线到 AiSession 和 process_tool_calls 由本 agent 完成。
---
### Agent 3:接口契约定义(跨 Wave 共享)
定义 Wave 2/Wave 3 依赖的数据结构,**仅定义不实现**,确保下游 agent 可独立编码。
```rust
// ── Wave 2 共享契约 ──
/// 压缩摘要双格式(T3 输出,T5 消费)
pub struct CompressedSummary {
pub json_card: String, // JSON 卡片(主题/决策/待办)
pub nl_summary: String, // 自然语言摘要(向前兼容)
}
// ── Wave 3 共享契约 ──
/// 工作流 DAG 上下文块(T4 输出 → 注入 system prompt
pub struct WorkflowContextBlock {
pub workflow_id: String,
pub workflow_name: String,
pub current_node: Option<String>,
pub completed: usize,
pub total: usize,
pub next_steps: Vec<String>,
}
/// WorkingContext 常驻上下文(T5 实现)
pub struct WorkingContext {
pub goals: Vec<GoalItem>,
pub decisions: Vec<DecisionItem>,
pub unresolved: Vec<String>,
pub current_step: Option<StepInfo>,
pub version: WorkingContextVersion,
}
pub enum GoalStatus { Active, Completed, Cancelled }
pub struct GoalItem { pub id: String, pub text: String, pub status: GoalStatus }
pub struct DecisionItem { pub question: String, pub decision: String, pub reason: String }
pub struct StepInfo { pub step_name: String, pub status: StepStatus }
pub enum StepStatus { Pending, InProgress, Done, Blocked }
pub struct WorkingContextVersion {
pub goals_changed_at: u32,
pub decisions_changed_at: u32,
pub unresolved_changed_at: u32,
pub step_changed_at: u32,
pub current_turn: u32,
}
```
**验证**`cargo check` — 编译器验证所有 use 路径和类型引用正确,无需运行时测试。
**简化**:Agent 3 先把上述结构体定义放进 `crates/df-ai/src/context_helpers.rs``pub use` 重导出。Wave 2/3 的 agent 直接引用,不用等。
---
## Gate 1Wave 1 → Wave 2
```
条件:
□ cargo check --workspace 全部通过
□ cargo test -p df-ai 全部通过
□ cargo test -p devflow 全部通过(已有测试不受影响)
验证项:
□ 日志: compress 触发时 available 减去 sys_tokensT1
□ 手动: 大工具结果 DB 存原文(T2)
□ 手动: namespace 引用格式正确(T2
回滚:
单文件回退: T1 删参数,T2 删 `namespace_store.rs` + 恢复 process_tool_calls 原文 push
```
---
## Wave 2:结构化改进(3 agent 并行)
### Agent 4T3 L3 结构化摘要
**前置依赖**Agent 3 定义了 `CompressedSummary` 结构
**改动清单**
| 文件 | 行 | 操作 |
|------|----|------|
| `src-tauri/src/commands/ai/compress.rs` | 45-52 签名 | `compress_via_llm` 返回类型 `Result<String>``Result<CompressedSummary>` |
| 同上 | 82-86 prompt 构造 | 改 `compress_prompt` 模板要求输出 JSON+NL 双格式 |
| 同上 | 114-119 clean + 返回 | `clean_summary` 适配双格式:提取 JSON 段 + NL 段,分别验证退化 |
| 同上 | 新增 | `extract_between` 辅助函数(提取 `<<<JSON>>>...<<<END_JSON>>>` 之间的内容) |
| `src-tauri/src/commands/ai/prompt.rs` | 240-290 | `compress_prompt` 模板新增 JSON+NL 双格式输出要求 |
| `src-tauri/src/commands/ai/agentic/context_lifecycle.rs` | 145-155 消费处 | `Ok(Some(summary))` 分支中 `insert_at` 内容改为双格式嵌入 |
| 同上 178-210 兜底 | 关键词摘要降级 | 关键词降级不改,仍是纯文本(`keyword_fallback` 无 JSON |
**验证**
1. `cargo test -p devflow` — compress 单测通过
2. 手动:触发压缩 → DB 中 system 消息含 `<<<STRUCTURED_SUMMARY>>>` 标记
3. 手动:关键词摘要降级路径仍为纯文本(不受影响)
**与 Agent 5 无依赖**T3 只改 compress 路径,T4 只改 system prompt 注入,零耦合。
---
### Agent 5T4 工作流 DAG 注入
**前置依赖**Agent 3 定义了 `WorkflowContextBlock`Agent 2 的 namespace 接口已就绪(弱依赖——仅用于产出关联)
**改动清单**
| 文件 | 行 | 操作 |
|------|----|------|
| `src-tauri/src/commands/ai/workflow_context.rs` | 新增 | `load_dag()`, `build_active_path()`, `summarize_node()` |
| `src-tauri/src/commands/ai/agentic/mod.rs` | ~1110-1128 | `build_for_request` 之前:检测 `conv.workflow_id` → 加载 DAG → 构建 context block → 注入 system prompt |
| `crates/df-ai/src/context_helpers.rs` | 新增常量 | `WORKFLOW_DAG_MARKER` |
**核心逻辑**
```rust
// workflow_context.rs
pub fn build_active_path(dag: &Dag, current_node_id: &str) -> WorkflowContextBlock {
let path = dag.path_to_root(current_node_id); // 到根路径
let next = dag.children(current_node_id, 2); // 后续 2 层
WorkflowContextBlock {
workflow_id: dag.id.clone(),
workflow_name: dag.name.clone(),
total_nodes: dag.nodes.len(),
completed_nodes: dag.nodes.iter().filter(|n| n.status == "completed").count(),
current_node: summarize_node(dag.get_node(current_node_id)),
next_nodes: next.into_iter().map(|n| summarize_node(n)).collect(),
}
}
```
**关键安全**
```rust
// mod.rs: system_prompt 注入处
if let Some(wf_id) = &conv.workflow_id {
if let Ok(dag) = df_workflow::Dag::load(&db, wf_id) {
let block = build_active_path(&dag, &conv.current_node_id);
system_prompt = format!("[工作流] {}\n{}", block.to_system_text(), system_prompt);
}
// Dag::load 失败 → 静默跳过
}
// workflow_id == None → 行为完全不变
```
**验证**
1. `cargo check -p devflow` — 依赖 `df-workflow` crate 的 `Dag::load` 接口
2. 手动:创建带工作流的对话 → system prompt 含 `[工作流]`
3. 手动:自由对话(无 workflow)→ system prompt 不变
---
## Gate 2Wave 2 → Wave 3
```
条件:
□ cargo check --workspace 全部通过
□ cargo test --workspace 全部通过
□ 前端 vue-tsc --noEmit 通过
验证项:
□ 手动: 压缩后 DB system 消息含 JSON 标记(T3
□ 手动: 工作流对话注入 DAG 块(T4)
□ 手动: 自由对话不受影响(T4 安全)
□ 全量: history_tokens 在 T2 基础上再次下降(T3 摘要更精简)
回滚:
T3: 回退 compress.rs 签名 + prompt.rs 模板
T4: 删 workflow_context.rs + 恢复 mod.rs system_prompt 拼接
```
---
## Wave 3:常驻上下文(1 agent 串行)
### Agent 6T5 WorkingContext 分层注入
**前置依赖**
- Agent 3 的 `WorkingContext` 结构定义需在 `context_helpers.rs` 中(已在 Wave 1 Gate 前到位)
- Agent 4 的 `CompressedSummary` 用于 `reset_all_with_summary`(弱依赖——无 NL 摘要时可跳过)
**改动清单**
| 文件 | 行 | 操作 |
|------|----|------|
| `crates/df-ai/src/context_helpers.rs` | 已有定义 | `WorkingContext` / `WorkingContextVersion` 等已由 Agent 3 定义 |
| `crates/df-ai/src/context/mod.rs` | `ContextManager` struct | 新增字段 `pub working_context: Option<WorkingContext>` |
| `src-tauri/src/commands/ai/agentic/mod.rs` | ~960 system_prompt 拼接处 | 在 `build_for_request` 之前调 `WorkingContext::build_injection()` 注入 L1a+L1b |
| 同上 | tool_registry 注册 | 注册 `read_context` 工具(L1c |
| `src-tauri/src/commands/ai/agentic/context_lifecycle.rs` | ~145-155 压缩成功后 | 调 `ctx.version.reset_all_with_summary(&summary)` |
| `src-tauri/src/commands/ai/commands/chat.rs` | `ai_chat_clear_context` | 归档后 reset WorkingContext(归档 = 上下文状态变更) |
**脏标记核心逻辑**
```rust
impl WorkingContext {
pub fn build_injection(&self, recent_turns: u32) -> String {
let mut parts = vec![];
// L1a: 当前步骤(每轮必带,~200 tokens
if let Some(step) = &self.current_step {
parts.push(format!("当前步骤: {}", step.step_name));
}
// L1b: 过去 recent_turns 轮内有变更才注入
let threshold = self.version.current_turn.saturating_sub(recent_turns);
let has_active = self.goals.iter().any(|g| g.status == GoalStatus::Active);
if has_active && self.version.goals_changed_at >= threshold {
let active: Vec<_> = self.goals.iter()
.filter(|g| g.status == GoalStatus::Active)
.collect();
parts.push(format!("目标({}): {}", active.len(),
active.iter().map(|g| g.text.as_str()).collect::<Vec<_>>().join("; ")));
}
if self.version.unresolved_changed_at >= threshold && !self.unresolved.is_empty() {
parts.push(format!("待解: {}", self.unresolved.join("; ")));
}
parts.join("\n")
}
}
```
**验证**
1. `cargo test -p df-ai``WorkingContext::build_injection` 单元测试(脏标记正确跳过/注入)
2. 手动:观察 system prompt 含 L1a(当前步骤)和条件性的 L1b(有变更时)
3. 手动:压缩触发后,下轮 L1b 正常注入(`reset_all` 生效)
4. 开关:`GOAL_PIN_ENABLED=false`(或等价的 workincontext 开关)→ 行为恢复到改动前
---
## 汇总:执行序列
```
Week 1:
Day 1: Agent 3 定义契约(30 分钟) → 其余 3 agent 拿到接口开始编码
Day 1-2: Agent 1 T1 + Agent 2 T2 并行编码
Day 3: Gate 1 验证 + 修复
Week 2:
Day 4-5: Agent 4 T3 + Agent 5 T4 并行编码
Day 6: Gate 2 验证 + 修复
Week 3:
Day 7-9: Agent 6 T5 编码(由于串行依赖,此阶段无并行)
Day 10: 最终验证 + 全量回归测试
```
## 回滚总策略
| 场景 | 操作 |
|------|------|
| T1 有问题 | 删 `sys_tokens` 参数 + 恢复 `budget` 变量 → 2 行回退 |
| T2 有问题 | 删 `namespace_store.rs` + 恢复 `process_tool_calls` push 原文 → 1 文件回退 |
| T3 有问题 | 回退 `compress.rs` + `prompt.rs``Result<String>` → 2 文件回退 |
| T4 有问题 | 删 `workflow_context.rs` + 恢复 mod.rs system_prompt 拼接 → 1 文件回退 |
| T5 有问题 | `GOAL_PIN_ENABLED=false` 等效开关 → 配置回退,不改代码 |
| 全量回退 | `git revert <commit-hash>` |
+1
View File
@@ -62,6 +62,7 @@ docs/
│ ├── 意图识别层论证-2026-06-19.md # 通用前置意图识别层 8 维度论证 │ ├── 意图识别层论证-2026-06-19.md # 通用前置意图识别层 8 维度论证
│ ├── 上下文管理演进与发散思考-2026-07-20.md # 上下文管理现状评估、业界对标、7个发散方向与组合分析 │ ├── 上下文管理演进与发散思考-2026-07-20.md # 上下文管理现状评估、业界对标、7个发散方向与组合分析
│ ├── 上下文管理演进-任务技术设计-2026-07-20.md # T2命名空间/T4工作流DAG/T3结构化摘要技术设计与推演 │ ├── 上下文管理演进-任务技术设计-2026-07-20.md # T2命名空间/T4工作流DAG/T3结构化摘要技术设计与推演
│ ├── 上下文管理演进-执行计划-2026-07-20.md # Wave 划分、agent 并行策略、接口契约、验证 gate、回滚策略
│ ├── 多主题上下文管理愿景-2026-06-19.md # 多主题多摘要愿景(远期方向,关联 F-15/意图识别) │ ├── 多主题上下文管理愿景-2026-06-19.md # 多主题多摘要愿景(远期方向,关联 F-15/意图识别)
│ └── 查询效率优化方案-2026-06-19.md # 查询链路优化(SQL 下推/精确拉取/缓存/字段投影) │ └── 查询效率优化方案-2026-06-19.md # 查询链路优化(SQL 下推/精确拉取/缓存/字段投影)
├── 03-模块文档/ ├── 03-模块文档/
@@ -69,6 +69,7 @@ pub(super) async fn maybe_auto_compress(
provider_config: &AiProviderRecord, provider_config: &AiProviderRecord,
llm_concurrency: &LlmConcurrency, llm_concurrency: &LlmConcurrency,
iteration: usize, iteration: usize,
sys_tokens: u32,
) -> bool { ) -> bool {
let prev_compressing = session_arc.lock().await.conv_read(&conv_id).map(|c| c.messages.is_compressing()).unwrap_or(false); let prev_compressing = session_arc.lock().await.conv_read(&conv_id).map(|c| c.messages.is_compressing()).unwrap_or(false);
if !prev_compressing { if !prev_compressing {
@@ -85,10 +86,10 @@ pub(super) async fn maybe_auto_compress(
let mgr = &mgr.messages; let mgr = &mgr.messages;
let protect_start = mgr.len().saturating_sub(PROTECT_COUNT); let protect_start = mgr.len().saturating_sub(PROTECT_COUNT);
let history_tokens = mgr.history_tokens(); let history_tokens = mgr.history_tokens();
let budget = mgr.budget_limit(); let available = mgr.budget_limit().saturating_sub(sys_tokens);
// 触发阈值 0.6*budget(整数比避免浮点):budget*6/10 < history_tokens // 触发阈值 0.6*available(可用预算=budget-sys_tokens):available*6/10 < history_tokens
let should = protect_start > 0 let should = protect_start > 0
&& (budget as u64) * 6 / 10 < history_tokens as u64 && (available as u64) * 6 / 10 < history_tokens as u64
&& mgr.has_compressible_messages(protect_start); && mgr.has_compressible_messages(protect_start);
(should, protect_start, history_tokens) (should, protect_start, history_tokens)
}; };
+1
View File
@@ -1122,6 +1122,7 @@ pub(crate) async fn run_agentic_loop(
&provider_config, &provider_config,
&llm_concurrency, &llm_concurrency,
iteration, iteration,
sys_tokens,
).await { ).await {
return; return;
} }
+14 -2
View File
@@ -486,7 +486,13 @@ pub(crate) async fn process_tool_calls(
// 短 lock 段:push tool_result(纯写,无 await) // 短 lock 段:push tool_result(纯写,无 await)
{ {
let mut session = session_arc.lock().await; let mut session = session_arc.lock().await;
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, content.clone())); // T2: 大工具结果进 namespace
let msg_content = if df_ai::namespace_store::should_use_namespace(&content, &draft.name) {
session.namespace_store.store(&draft.name, &content)
} else {
content.clone()
};
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &msg_content));
} }
// 锁外:审计(trust 放行仍记一条,decided_by=auto_trust),留痕可追溯 // 锁外:审计(trust 放行仍记一条,decided_by=auto_trust),留痕可追溯
audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, status, risk_level, Some(content), Some("auto_trust"), current_message_id).await; audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, status, risk_level, Some(content), Some("auto_trust"), current_message_id).await;
@@ -599,7 +605,13 @@ pub(crate) async fn process_tool_calls(
// 短 lock 段:写 per_conv.messages(纯写,无 await) // 短 lock 段:写 per_conv.messages(纯写,无 await)
{ {
let mut session = session_arc.lock().await; let mut session = session_arc.lock().await;
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, content.clone())); // T2: 大工具结果进 namespace,主队列只留引用
let msg_content = if df_ai::namespace_store::should_use_namespace(&content, &draft.name) {
session.namespace_store.store(&draft.name, &content)
} else {
content.clone()
};
session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &msg_content));
} }
// F-#97 审计留痕:low_risk 向量中 risk_level != Low 即 mode 放行(Med 仅 medium/all、High 仅 all 才进)。 // F-#97 审计留痕:low_risk 向量中 risk_level != Low 即 mode 放行(Med 仅 medium/all、High 仅 all 才进)。
// decided_by 区分接管来源,审计表可追溯 all 模式执行了多少高危命令(治 securityReview blocker: // decided_by 区分接管来源,审计表可追溯 all 模式执行了多少高危命令(治 securityReview blocker:
+11
View File
@@ -222,6 +222,17 @@ async fn save_conversation_inner(
eprintln!("[LOCK-DIAG] save clone 段持锁 {:?} (含等待,>50ms 报告,定位 session lock 长持有者)", __lock_total); eprintln!("[LOCK-DIAG] save clone 段持锁 {:?} (含等待,>50ms 报告,定位 session lock 长持有者)", __lock_total);
} }
// truncate 在锁外(clone 副本上操作,不影响 session 真相源) // truncate 在锁外(clone 副本上操作,不影响 session 真相源)
// T2: 展开 namespace 引用为原文再落库
{
let session = session_arc.lock().await;
for m in &mut msgs {
if df_ai::namespace_store::is_namespace_ref(&m.content) {
if let Some(original) = session.namespace_store.read_only(&m.content) {
m.content = original.to_string();
}
}
}
}
for m in &mut msgs { for m in &mut msgs {
// P0-2:tool result 是结构化 JSON(provider 读工具返回原样), // P0-2:tool result 是结构化 JSON(provider 读工具返回原样),
// 中段截断会插裸换行+中文省略标记破坏 JSON 结构致 reload/重发 parse FAIL。 // 中段截断会插裸换行+中文省略标记破坏 JSON 结构致 reload/重发 parse FAIL。
+3
View File
@@ -530,6 +530,8 @@ pub struct AiSession {
/// conversation.rs save + title.rs + knowledge_inject.rs + lib.rs L0)读写 /// conversation.rs save + title.rs + knowledge_inject.rs + lib.rs L0)读写
/// `session.conv(conv_id).*` / `session.conv_read(conv_id)`。顶层单例会话级字段已全部删除。 /// `session.conv(conv_id).*` / `session.conv_read(conv_id)`。顶层单例会话级字段已全部删除。
pub per_conv: HashMap<String, PerConvState>, pub per_conv: HashMap<String, PerConvState>,
/// T2 工具命名空间存储(运行时缓存,大工具结果不进主队列)
pub namespace_store: df_ai::namespace_store::NamespaceStore,
} }
impl AiSession { impl AiSession {
@@ -540,6 +542,7 @@ impl AiSession {
active_conv_created_at: None, active_conv_created_at: None,
pending_approvals: HashMap::new(), pending_approvals: HashMap::new(),
per_conv: HashMap::new(), per_conv: HashMap::new(),
namespace_store: df_ai::namespace_store::NamespaceStore::default(),
} }
} }