Compare commits

...

10 Commits

Author SHA1 Message Date
oudecheng
4baa8e7a6b fix: 内存历史按 topic_id 键化,新增 replace_topic_history 修复 DB 压缩覆盖
- SessionHistory.chat_histories 改为 topic_histories 按 topic_id 键化,消除多话题并发时内存历史互相覆盖的根因

- is_current_turn 改用 current_topic == original_topic_id 直接比较,替代失效的内存最新消息匹配判断

- 新增 ConversationRepository::replace_topic_history 按 topic 删除/插入消息,解决 replace_active_history 删除整个 session 消息的 pre-existing 问题

- compaction 路径改用 replace_topic_history,避免压缩时覆盖其他 topic 的 DB 消息

- switch_topic 不再 remove_history,不同 topic 历史独立存储互斥

- append_persisted_message 仅当 topic_id == current_topic 时更新内存,防止延迟消息污染新话题历史
2026-07-28 15:32:42 +08:00
oudecheng
b042b45ac7 fix: 话题隔离 - topic_id 全程显式传递,修复消息错投与并发阻塞
问题:WS 端一个话题执行中时新建另一话题发消息,新会话无响应
(串行锁按 chat_id 阻塞),且首个话题完成后用户消息错误出现在
旧话题而非新话题(执行路径多次从共享 UI 状态读取 topic_id 产生竞态)。

核心修复(第一性原则):
执行上下文应在消息接收时一次性捕获,全程显式传递,不从共享可变
状态重复读取。

1. processor.rs: process_one 入口捕获 current_topic,传入 handle_message
   和 set_agent_cancel_token
2. session_message_service.rs: handle_message 签名加 topic_id 参数,
   透传给 MessageExecutionRequest
3. execution.rs: MessageExecutionRequest 加 topic_id 字段;
   - 串行锁键改用 topic_id(不同 topic 并发,同 topic 串行)
   - original_topic_id 优先用传入值,消除锁等待期间 topic 切换竞态
   - append_persisted_message 调用传入 original_topic_id
   - create_agent 调用传入 original_topic_id
4. session.rs: append_persisted_message 加 explicit_topic_id 参数;
   create_agent/create_agent_with_provider_config 加 explicit_topic_id;
   set_cancel_receiver/set_agent_cancel_token 加 topic_id 参数;
   pending_cancel_tokens 查找改为优先 topic_id(避免并发 topic 执行时
   cancel token 互相覆盖)

对抗性审查补丁:
- append_persisted_message: 仅当写入 topic 匹配当前活跃 topic 时才更新
  内存历史,避免旧 topic 的排队消息污染已切换到的新 topic 内存历史
- prepare_and_execute_scheduled_task: 锁前一次性捕获 topic_id,锁后
  复用同一值作为 original_topic_id,保证锁键与写入目标一致

已验证:cargo check 通过,gateway 模块 48 个测试通过(1 个预存在的
prompt 模板测试失败,与本次修改无关)。
2026-07-28 14:19:20 +08:00
oudecheng
3a8da51936 refactor: 串行锁按 topic_id 键化,支持多话题并发执行
将 chat_serial_locks 重命名为 topic_serial_locks,串行锁粒度从
chat_id 改为 topic_id。同一 topic 的消息处理仍串行执行,不同
topic 之间互不阻塞,为后续多话题并发修复铺路。

本提交仅重命名锁结构和访问方法,调用点仍用 chat_id 作为锁键
(回退兼容),不影响现有行为。
2026-07-28 14:17:12 +08:00
56612389ae 改用gray_matter解析md头元数据,增加相应的日志 2026-07-21 16:33:55 +08:00
oudecheng
141ffda1ee feat: 实现聊天消息的串行锁,确保同一聊天的消息处理串行执行 2026-07-15 17:47:52 +08:00
oudecheng
303f6d83e3 feat: 添加工具调用序列的前向检查,确保工具消息紧随助手消息后 2026-07-15 16:36:13 +08:00
oudecheng
cde41e32a8 refactor: 优化工具使用说明,简化记忆检索和写入规则 2026-07-13 17:00:40 +08:00
oudecheng
cea4bd3cfb feat: 子代理支持数据驱动的工具过滤(白名单+黑名单),删除内置 explore
SUBAGENT.md frontmatter 新增 denied_tools 黑名单字段,启用已有的 allowed_tools 白名单;新增 ToolRegistry::only 白名单方法;抽取 build_subagent_tools_registry(白名单→黑名单→depth 兜底);安全修复:resume 在 def 失踪时拒绝恢复而非降级为完整工具集(避免权限提升);depth 阈值改为引用 max_nesting_depth 配置;SubagentWithStatus 暴露工具字段供前端只读展示;删除内置 explore 子代理及专属 explore_max_execution_secs 配置(全栈清理);新增 23 个测试覆盖过滤矩阵、frontmatter 解析、状态投影
2026-07-09 14:42:38 +08:00
oudecheng
c276381d6c 增加整理意识,确保同类工作输出到同一文件夹 2026-07-08 16:32:43 +08:00
oudecheng
58f461c953 fix: 修复 SSRF 重定向绕过 + 符号链接路径遍历 + TodoItemSummary 字段缺失
P0: src/tools/http_request.rs SSRF 重定向绕过
  - reqwest::Client 默认跟随最多 10 次重定向,is_private_host 仅检查初始 URL
  - 攻击者可用公网 URL 返回 302 → http://127.0.0.1/http://169.254.169.254/(云元数据端点)绕过防护访问内网
  - 修复:.redirect(reqwest::redirect::Policy::none()) 完全禁用重定向

P1: src/tools/file_read/write/edit.rs 符号链接路径遍历
  - resolve_path 用 starts_with 检查但未 canonicalize
  - 攻击者可在 allowed_dir 内创建指向 /etc/passwd 的符号链接绕过限制
  - 修复:对 resolved 和 allowed 均执行 canonicalize 后比较
  - file_read: 文件必须存在,canonicalize 失败直接报错
  - file_write/edit: 文件可能不存在,降级到父目录 canonicalize

P1: src/protocol/mod.rs + list_todos.rs TodoItemSummary 字段缺失
  - 后端 TodoItemSummary 仅返回 4 字段,前端期望 7 字段
  - 缺失 priority, created_at, updated_at,前端 TodoPanel 无法显示
    优先级和时间戳
  - 修复:struct 补齐 3 字段,list_todos 构造时传递完整字段
2026-07-08 16:23:09 +08:00
35 changed files with 1667 additions and 575 deletions

82
Cargo.lock generated
View File

@ -19,6 +19,18 @@ dependencies = [
"cpufeatures 0.2.17",
]
[[package]]
name = "ahash"
version = "0.8.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
dependencies = [
"cfg-if",
"once_cell",
"version_check",
"zerocopy",
]
[[package]]
name = "aho-corasick"
version = "1.1.4"
@ -28,6 +40,12 @@ dependencies = [
"memchr",
]
[[package]]
name = "allocator-api2"
version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
[[package]]
name = "android_system_properties"
version = "0.1.5"
@ -93,6 +111,12 @@ version = "1.0.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
[[package]]
name = "arraydeque"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236"
[[package]]
name = "async-trait"
version = "0.1.89"
@ -545,7 +569,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users 0.5.2",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -616,7 +640,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -844,6 +868,27 @@ dependencies = [
"weezl",
]
[[package]]
name = "gray_matter"
version = "0.2.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8666976c40b8633f918783969b6681a3ddb205f29150348617de425d85a3e3bd"
dependencies = [
"serde",
"serde_json",
"yaml-rust2",
]
[[package]]
name = "hashbrown"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
dependencies = [
"ahash",
"allocator-api2",
]
[[package]]
name = "hashbrown"
version = "0.15.5"
@ -868,6 +913,15 @@ version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
[[package]]
name = "hashlink"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7"
dependencies = [
"hashbrown 0.14.5",
]
[[package]]
name = "hashlink"
version = "0.11.1"
@ -1485,7 +1539,7 @@ version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -1596,6 +1650,7 @@ dependencies = [
"dotenv",
"encoding_rs",
"futures-util",
"gray_matter",
"http",
"iana-time-zone",
"image",
@ -2123,7 +2178,7 @@ dependencies = [
"bitflags",
"fallible-iterator",
"fallible-streaming-iterator",
"hashlink",
"hashlink 0.11.1",
"libsqlite3-sys",
"smallvec",
"sqlite-wasm-rs",
@ -2188,7 +2243,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -2247,7 +2302,7 @@ dependencies = [
"security-framework",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -2656,7 +2711,7 @@ dependencies = [
"getrandom 0.4.2",
"once_cell",
"rustix",
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -3347,7 +3402,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.59.0",
"windows-sys 0.61.2",
]
[[package]]
@ -3657,6 +3712,17 @@ version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
[[package]]
name = "yaml-rust2"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8902160c4e6f2fb145dbe9d6760a75e3c9522d8bf796ed7047c85919ac7115f8"
dependencies = [
"arraydeque",
"encoding_rs",
"hashlink 0.8.4",
]
[[package]]
name = "yoke"
version = "0.8.2"

View File

@ -41,6 +41,7 @@ rustls = { version = "0.23", features = ["ring"] }
wechatbot = { path = "vendor/wechatbot" }
encoding_rs = "0.8"
libc = "0.2"
gray_matter = { version = "0.2", default-features = false, features = ["yaml"] }
[target.'cfg(windows)'.dependencies]
windows-sys = { version = "0.59", features = [

View File

@ -498,7 +498,7 @@ tools 配置示例:
- shell - 执行 shell 命令Windows PowerShell/Cmd
- http_request - HTTP 请求
- web_fetch - 网页抓取
- task - 创建和管理子代理支持内置类型general/explore)和用户自定义类型
- task - 创建和管理子代理支持内置类型general和用户自定义类型
注意bash 和 shell 是同一个工具在不同平台上的名称,运行时自动检测。
@ -509,7 +509,8 @@ PicoBot 支持通过 `task` 工具创建子代理来处理复杂多步骤任务
### 8.1 内置子代理类型
- **general**通用型子代理适合处理复杂多步骤任务。可以使用读写文件、执行命令、HTTP 请求等完整工具集。
- **explore**:探索型子代理,用于代码库探索和信息收集。只使用只读工具,禁止任何写操作。
> 如需只读探索型子代理,可通过自定义子代理配合 `allowed_tools` 白名单实现(见下文)。
### 8.2 自定义子代理
@ -538,7 +539,8 @@ prompt_template: |
3. 完成后给出简洁的总结
注意: 你是一个只读代理,禁止执行任何修改操作。
allowed_tools: [read, bash, web_fetch] # 可选,覆盖默认工具白名单
allowed_tools: [read, bash, web_fetch] # 可选,工具白名单(仅这些工具可用)
denied_tools: [task] # 可选,工具黑名单(这些工具被禁用)
max_execution_secs: 600 # 可选,覆盖默认执行时间
---
@ -555,9 +557,12 @@ max_execution_secs: 600 # 可选,覆盖默认执行时间
| `name` | string | 否 | 子代理名称,默认取目录名 |
| `description` | string | 是 | 简短描述,用于 agent 选择 |
| `prompt_template` | string | 是 | 提示词模板,支持变量插值 |
| `allowed_tools` | array | 否 | 工具白名单,不指定时使用默认列表 |
| `allowed_tools` | array | 否 | 工具白名单,指定后仅这些工具可用;不指定则不限制 |
| `denied_tools` | array | 否 | 工具黑名单,指定后这些工具被禁用;在白名单之后应用 |
| `max_execution_secs` | integer | 否 | 最大执行时间(秒) |
> **工具过滤语义**`allowed_tools``denied_tools` 可共存。生效顺序为:先应用白名单(取交集),再扣除黑名单。两者都不指定时,子代理使用完整工具集。当子代理嵌套深度达到 `max_nesting_depth`(默认 2即孙代理始终移除 `task` 工具以防无限嵌套。白名单中未注册的工具名会被静默跳过。
#### 模板变量
`prompt_template` 支持以下变量插值:
@ -615,7 +620,7 @@ PicoBot 的 Agent 是围绕工具调用构建的。当前默认注册的工具
- bash / shell执行 shell 命令同一工具Unix 下名称为 bashWindows 下名称为 shell
- http_request发起 HTTP 请求
- web_fetch抓取网页正文
- task创建和管理子代理支持内置类型general/explore)和用户自定义类型
- task创建和管理子代理支持内置类型general和用户自定义类型
其中:
@ -625,7 +630,7 @@ PicoBot 的 Agent 是围绕工具调用构建的。当前默认注册的工具
- skill_activate 负责把具体技能正文注入当前任务上下文
- skill_manage 整合了技能列出与管理功能,支持运行时创建、更新、删除和批量禁用
- bash / shell / http_request / web_fetch 让 Agent 具备更强的外部交互能力bash 和 shell 是同一工具在不同平台的名称)
- task 允许 Agent 创建独立上下文的子代理来处理复杂多步骤任务支持内置类型general/explore)和用户自定义类型
- task 允许 Agent 创建独立上下文的子代理来处理复杂多步骤任务支持内置类型general和用户自定义类型
### 9.1 MCP 工具集成

View File

@ -2453,6 +2453,57 @@ mod tests {
// Verify no tool messages remain
assert!(messages.iter().all(|m| m.role != "tool"));
}
#[test]
fn test_sanitize_strips_tool_calls_when_tool_results_not_immediately_following() {
// [assistant(tool_calls=[A]), user, tool(A)]
// Tool result exists but is NOT immediately after assistant → strip tool_calls
// This is the scenario that triggers DeepSeek 400 "insufficient tool messages
// following tool_calls message" — the reverse scan sees tool(A) as resolved,
// but the API requires it to be IMMEDIATELY after the assistant.
let mut messages = vec![
ChatMessage::assistant_with_tool_calls(
"calling tool",
vec![ToolCall {
id: "call_A".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({}),
}],
),
ChatMessage::user("interrupting message"),
ChatMessage::tool("call_A", "search", "result"),
];
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages);
// The assistant should be removed (tool_calls stripped via removal)
// and the orphaned tool(A) should also be removed
assert!(removed >= 2, "should remove both the assistant and orphaned tool message, got {}", removed);
assert_eq!(messages.len(), 1, "only the user message should remain");
assert_eq!(messages[0].role, "user");
assert!(messages.iter().all(|m| m.tool_calls.as_ref().map_or(true, |c| c.is_empty())),
"no assistant should have tool_calls remaining");
}
#[test]
fn test_sanitize_preserves_tool_calls_when_immediately_followed() {
// [assistant(tool_calls=[A]), tool(A), user] — valid, tool result is immediate
let mut messages = vec![
ChatMessage::assistant_with_tool_calls(
"calling tool",
vec![ToolCall {
id: "call_A".to_string(),
name: "search".to_string(),
arguments: serde_json::json!({}),
}],
),
ChatMessage::tool("call_A", "search", "result"),
ChatMessage::user("next message after tool result"),
];
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages);
assert_eq!(removed, 0, "should not remove anything — tool result immediately follows");
assert_eq!(messages.len(), 3);
}
}
#[derive(Debug)]

View File

@ -316,6 +316,86 @@ pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMes
}
}
// Phase 1.5: Forward-order check — verify tool messages IMMEDIATELY follow
// the assistant(tool_calls). If any non-tool message appears between the
// assistant and its tool results, the API rejects with
// "insufficient tool messages following tool_calls message".
//
// The reverse scan in Phase 1 only checks existence (tool result appears
// somewhere after assistant), NOT immediacy. This pass catches cases like:
// [assistant(tool_calls=[A]), user, tool(A)]
// ^ Phase 1 sees tool(A) after assistant → "resolved"
// but API requires tool(A) to be IMMEDIATELY after assistant
{
let mut pending_tool_ids: HashSet<String> = HashSet::new();
let mut pending_assistant_idx: Option<usize> = None;
for (i, m) in messages.iter().enumerate() {
// If we have pending tool_ids and encounter a non-tool message,
// the assistant's tool results were NOT immediately following.
if !pending_tool_ids.is_empty() && m.role != "tool" {
if let Some(idx) = pending_assistant_idx {
if !remove_indices.contains(&idx) {
tracing::warn!(
message_index = idx,
interrupted_by_index = i,
interrupted_by_role = %m.role,
pending_tool_call_count = pending_tool_ids.len(),
"Removing assistant with tool_calls — tool results \
not immediately following (interrupted by non-tool message)"
);
// Remove this assistant's tool_call_ids from with_parent
// so Phase 2 cleans up the now-orphaned tool messages
if let Some(calls) = messages[idx].tool_calls.as_ref() {
for tc in calls.iter() {
with_parent.remove(&tc.id);
}
}
remove_indices.push(idx);
}
}
pending_tool_ids.clear();
pending_assistant_idx = None;
}
if m.role == "assistant"
&& m.tool_calls.as_ref().map_or(false, |calls| !calls.is_empty())
{
let already_marked = remove_indices.contains(&i);
if !already_marked {
pending_tool_ids = m.tool_calls.as_ref().unwrap()
.iter().map(|tc| tc.id.clone()).collect();
pending_assistant_idx = Some(i);
}
} else if m.role == "tool" {
if let Some(ref tc_id) = m.tool_call_id {
pending_tool_ids.remove(tc_id);
if pending_tool_ids.is_empty() {
pending_assistant_idx = None;
}
}
}
}
// Handle trailing assistant with unresolved immediate tool results
if !pending_tool_ids.is_empty() {
if let Some(idx) = pending_assistant_idx {
if !remove_indices.contains(&idx) {
tracing::warn!(
message_index = idx,
"Removing trailing assistant with incomplete immediate tool results"
);
if let Some(calls) = messages[idx].tool_calls.as_ref() {
for tc in calls.iter() {
with_parent.remove(&tc.id);
}
}
remove_indices.push(idx);
}
}
}
}
// Remove in descending index order to avoid shifting
for &idx in &remove_indices {
messages.remove(idx);

View File

@ -76,6 +76,9 @@ impl CommandHandler for ListTodosCommandHandler {
id: r.id,
content: r.content,
status: r.status,
priority: r.priority,
created_at: r.created_at,
updated_at: r.updated_at,
created_by_message_id: r.created_by_message_id,
})
.collect();

View File

@ -248,8 +248,6 @@ pub struct TaskConfig {
pub enabled: bool,
#[serde(default = "default_task_max_execution_secs")]
pub max_execution_secs: u64,
#[serde(default = "default_task_explore_max_execution_secs")]
pub explore_max_execution_secs: u64,
#[serde(default = "default_task_ttl_hours")]
pub ttl_hours: u64,
#[serde(default = "default_task_allowed_tools")]
@ -266,10 +264,6 @@ fn default_task_max_execution_secs() -> u64 {
3600 // 60分钟
}
fn default_task_explore_max_execution_secs() -> u64 {
3600 // 60分钟
}
fn default_task_ttl_hours() -> u64 {
24
}
@ -300,7 +294,6 @@ impl Default for TaskConfig {
Self {
enabled: default_task_enabled(),
max_execution_secs: default_task_max_execution_secs(),
explore_max_execution_secs: default_task_explore_max_execution_secs(),
ttl_hours: default_task_ttl_hours(),
allowed_tools: default_task_allowed_tools(),
max_nesting_depth: default_task_max_nesting_depth(),

View File

@ -795,11 +795,14 @@ fn load_experts_from_root(root: &Path, source: ExpertSource) -> Vec<Expert> {
fn parse_expert_file(path: &Path, source: ExpertSource) -> Result<Expert, String> {
let content = fs::read_to_string(path).map_err(|e| format!("failed to read file: {}", e))?;
let (frontmatter_raw, body) =
split_frontmatter(&content).ok_or_else(|| "missing YAML frontmatter block".to_string())?;
let frontmatter: ExpertFrontmatter = serde_yaml::from_str(frontmatter_raw)
.map_err(|e| format!("invalid YAML frontmatter: {}", e))?;
let (frontmatter, body) = match crate::frontmatter::parse::<ExpertFrontmatter>(&content) {
Ok(v) => v,
Err(err) => {
let bytes = content.len();
let crlf = content.contains('\r');
return Err(format!("{} (bytes={}, crlf={})", err, bytes, crlf));
}
};
let description = frontmatter.description.trim();
if description.is_empty() {
@ -823,26 +826,6 @@ fn parse_expert_file(path: &Path, source: ExpertSource) -> Result<Expert, String
})
}
fn split_frontmatter(content: &str) -> Option<(&str, &str)> {
// 兼容 CRLFWindows和 LFUnix行尾符
let rest = content
.strip_prefix("---\n")
.or_else(|| content.strip_prefix("---\r\n"))?;
let marker = "\n---\n";
let marker_crlf = "\n---\r\n";
if let Some(idx) = rest.find(marker) {
let frontmatter = &rest[..idx];
let body = &rest[idx + marker.len()..];
Some((frontmatter, body))
} else if let Some(idx) = rest.find(marker_crlf) {
let frontmatter = &rest[..idx];
let body = &rest[idx + marker_crlf.len()..];
Some((frontmatter, body))
} else {
None
}
}
// ========== State file I/O ==========
fn load_expert_disable_state(cwd: &Path) -> ExpertDisableState {
@ -992,11 +975,17 @@ mod tests {
}
#[test]
fn test_split_frontmatter() {
let input = "---\ndescription: demo\n---\nhello";
let (fm, body) = split_frontmatter(input).unwrap();
assert!(fm.contains("description"));
assert_eq!(body, "hello");
fn test_parse_expert_file_handles_crlf_endings() {
let dir = tempfile::tempdir().unwrap();
let expert_dir = dir.path().join("demo");
fs::create_dir_all(&expert_dir).unwrap();
let expert_md = expert_dir.join("EXPERT.md");
fs::write(&expert_md, "---\r\ndescription: demo expert\r\n---\r\nStep A\r\nStep B").unwrap();
let expert = parse_expert_file(&expert_md, ExpertSource::Project).unwrap();
assert_eq!(expert.name, "demo");
assert_eq!(expert.description, "demo expert");
assert_eq!(expert.body, "Step A\nStep B");
}
#[test]

86
src/frontmatter.rs Normal file
View File

@ -0,0 +1,86 @@
use gray_matter::engine::YAML;
use gray_matter::Matter;
use serde::de::DeserializeOwned;
/// Parse a markdown document with YAML frontmatter into `(frontmatter, body)`.
///
/// Tolerates CRLF, CR, and LF line endings. A `---` appearing inside the body
/// is not treated as a delimiter (only the leading frontmatter block is split).
///
/// - Returns `Err("missing YAML frontmatter block")` when no leading `---`
/// delimiter is present.
/// - Returns `Err("invalid YAML frontmatter: {e}")` when the block is present
/// but the YAML fails to parse or deserialize into `T`.
pub fn parse<T: DeserializeOwned>(content: &str) -> Result<(T, String), String> {
let normalized = content.replace("\r\n", "\n").replace('\r', "\n");
if !normalized.starts_with("---\n") {
return Err("missing YAML frontmatter block".to_string());
}
let matter = Matter::<YAML>::new();
let result = matter.parse(&normalized);
let pod = result
.data
.ok_or_else(|| "invalid YAML frontmatter".to_string())?;
let data: T = pod
.deserialize()
.map_err(|e| format!("invalid YAML frontmatter: {}", e))?;
Ok((data, result.content))
}
#[cfg(test)]
mod tests {
use super::*;
use serde::Deserialize;
#[derive(Debug, Deserialize, PartialEq)]
struct FrontMatter {
description: String,
#[serde(default)]
name: Option<String>,
}
#[test]
fn parses_lf_endings() {
let input = "---\ndescription: demo\n---\nbody text";
let (fm, body) = parse::<FrontMatter>(input).unwrap();
assert_eq!(fm, FrontMatter { description: "demo".to_string(), name: None });
assert_eq!(body, "body text");
}
#[test]
fn parses_crlf_endings() {
let input = "---\r\ndescription: demo\r\n---\r\nbody text";
let (fm, body) = parse::<FrontMatter>(input).unwrap();
assert_eq!(fm.description, "demo");
assert_eq!(body, "body text");
}
#[test]
fn parses_cr_endings() {
let input = "---\rdescription: demo\r---\rbody text";
let (fm, body) = parse::<FrontMatter>(input).unwrap();
assert_eq!(fm.description, "demo");
assert_eq!(body, "body text");
}
#[test]
fn missing_block_is_rejected() {
let err = parse::<FrontMatter>("no front matter here").unwrap_err();
assert_eq!(err, "missing YAML frontmatter block");
}
#[test]
fn invalid_yaml_is_rejected() {
let err = parse::<FrontMatter>("---\n: : bad\n---\nbody").unwrap_err();
assert!(err.starts_with("invalid YAML frontmatter"));
}
#[test]
fn body_with_inner_delimiter_is_preserved() {
let input = "---\ndescription: demo\n---\nexcerpt\n---\nmore content";
let (_fm, body) = parse::<FrontMatter>(input).unwrap();
assert_eq!(body, "excerpt\n---\nmore content");
}
}

View File

@ -1,25 +0,0 @@
<!--
# 自定义 Agent 配置
在此文件添加您的个性化配置,这些内容会与系统默认提示词合并。
## 可用配置项示例
### 额外身份
- 我是后端开发者
- 我熟悉 Rust / Python
### 个人偏好
- 代码注释使用英文
- 优先使用中文回复技术概念
### 工作方式
- 复杂任务请先给出整体方案
- 性能优化建议优先
## 注意
- 此文件内容会追加在系统默认提示词之后
- 留空或删除内容则不会生效
- 使用 HTML 注释 <!-- --> 可以保留说明文字而不生效
- 删除本注释块并添加您的自定义配置即可生效
-->

View File

@ -13,17 +13,21 @@ use super::session::Session;
/// has already finished by this point there is no response-time impact, and
/// the synchronous guarantee means the next execution always starts with
/// freshly compacted history.
///
/// 按 topic_id 隔离:压缩只处理指定 topic 的历史DB 替换也只影响该 topic。
pub(crate) async fn schedule_background_history_compaction(
session: Arc<Mutex<Session>>,
chat_id: impl Into<String>,
topic_id: impl Into<String>,
) -> Result<(), AgentError> {
let chat_id = chat_id.into();
let topic_id = topic_id.into();
let mut session_guard = session.lock().await;
session_guard.ensure_persistent_session(&chat_id)?;
session_guard.ensure_chat_loaded(&chat_id)?;
session_guard.ensure_chat_loaded(&chat_id, Some(&topic_id))?;
let history = session_guard.get_or_create_history(&chat_id).clone();
let history = session_guard.get_or_create_history(&topic_id).clone();
let compressor = session_guard.compressor().clone();
if !compressor.should_compress(&history) {
@ -36,6 +40,7 @@ pub(crate) async fn schedule_background_history_compaction(
tracing::info!(
chat_id = %chat_id,
topic_id = %topic_id,
msg_count = history.len(),
"Starting synchronous two-segment compression"
);
@ -47,19 +52,20 @@ pub(crate) async fn schedule_background_history_compaction(
.compress_two_segment(&history, &provider_config)
.await?;
// Replace the entire history with the compressed result.
// Since we hold the lock, no concurrent modifications can occur.
// Replace only this topic's history in DB (not the entire session).
// This avoids clobbering other topics' messages during compaction.
store
.replace_active_history(&session_id, &compressed)
.map_err(|e| AgentError::Other(format!("replace_active_history error: {}", e)))?;
.replace_topic_history(&session_id, &topic_id, &compressed)
.map_err(|e| AgentError::Other(format!("replace_topic_history error: {}", e)))?;
tracing::info!(
chat_id = %chat_id,
topic_id = %topic_id,
compressed_msg_count = compressed.len(),
"Two-segment compression committed"
);
session_guard.reload_chat_history(&chat_id)?;
session_guard.reload_topic_history(&chat_id, &topic_id)?;
Ok(())
}

View File

@ -21,6 +21,7 @@
- 复杂任务先收敛重点,简单任务直接给结果。
- 避免不必要的重复、客套和冗长说明。
- 调用工具的时候需要不仅仅回复工具的json也简短说明你调用工具要完成什么工作
- 具备良好的整理意识,同一类的工作输出放到同一个输出文件夹,而不是输出在根目录
## 回复规则
@ -32,10 +33,9 @@
- 回答应以帮助用户完成当前目标为中心。
- 在信息不足时先补关键前提,在信息充分时直接执行。
- 调用工具的时候必须同时用简短的话告诉用户你调用工具是做什么
- 无需担心创建子智能体过多的问题请按用户或者skill的要求创建对应数量的子智能体这样可以隔离上下文更好完成工作
- 思考的时候建议用中文思考
- 涉及到时间的都用get_time工具获取避免时间不准确
## 用户附件
用户发过来了一些附件先判断文件后缀名能不能直接读取如果不能直接read的比如xlsx,就要通过代码等其他方式读取里面的内容

View File

@ -64,6 +64,8 @@ pub(crate) struct MessageExecutionRequest<'a> {
pub(crate) content: &'a str,
pub(crate) media: Vec<MediaItem>,
pub(crate) live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
/// 消息接收时捕获的 topic_id全程显式传递避免从共享状态重复读取竞态
pub(crate) topic_id: Option<String>,
}
pub(crate) struct ScheduledExecutionRequest<'a> {
@ -89,13 +91,19 @@ impl AgentExecutionService {
session: &mut Session,
request: FinalizeAgentResultRequest<'_>,
) -> Result<FinalizedAgentResult, AgentError> {
// 检查是否是最新的用户回合
let is_current_turn =
session.matches_current_user_turn(request.chat_id, request.user_message);
// 判断是否是最新的用户回合
// 直接比较 current_topic(chat_id) 与 original_topic_id
// 这比"检查内存历史最新消息"更可靠,且天然处理"切走又切回"的 case
let is_current_turn = match request.original_topic_id.as_deref() {
Some(orig_tid) => session.current_topic(request.chat_id).as_deref() == Some(orig_tid),
None => true, // 无 topic 时总是视为当前回合
};
if !is_current_turn {
let (latest_user_id, latest_user_preview, compression_in_flight, history_len) =
session.stale_result_diagnostics(request.chat_id);
session.stale_result_diagnostics(
request.original_topic_id.as_deref().unwrap_or(request.chat_id),
);
tracing::info!(
channel = %request.channel_name,
chat_id = %request.chat_id,
@ -111,29 +119,25 @@ impl AgentExecutionService {
}
// 确定保存消息的话题 ID
// 如果是最新回合,使用当前话题;否则使用原始话题
let target_topic_id = if is_current_turn {
session.current_topic(request.chat_id)
} else {
request.original_topic_id.as_deref()
};
// 始终使用执行开始时捕获的 original_topic_id避免从共享状态重复读取竞态
let target_topic_id = request.original_topic_id.as_deref();
// 将结果消息保存到确定的话题
if let Some(topic_id) = target_topic_id {
if is_current_turn {
// 如果是最新回合,使用 append_persisted_messages 保存到数据库并更新内存历史
// 话题未切换current_topic == original_topic_id安全更新内存历史
if let Err(err) = session.append_persisted_messages(
request.chat_id,
topic_id,
request.result.emitted_messages.clone(),
) {
tracing::error!(
error = %err,
chat_id = %request.chat_id,
topic_id = %topic_id,
"Failed to append messages to session history"
);
}
} else {
// 如果用户已切换话题,只保存到原始话题(不更新内存历史)
// 话题已切换,只写 DB 不更新内存(避免污染新话题的历史)
if let Err(err) = session.append_messages_to_topic(
request.chat_id,
topic_id,
@ -147,7 +151,8 @@ impl AgentExecutionService {
}
}
} else if is_current_turn {
// 如果没有话题直接更新内存历史append_persisted_messages 会处理持久化)
// 没有话题直接更新内存历史append_persisted_messages 会处理持久化)
// 无 topic 场景用 chat_id 作为 topic_histories 的回退 key
if let Err(err) = session.append_persisted_messages(
request.chat_id,
request.result.emitted_messages.clone(),
@ -200,11 +205,33 @@ impl AgentExecutionService {
&self,
request: MessageExecutionRequest<'_>,
) -> Result<Vec<OutboundMessage>, AgentError> {
// 获取该 topic 的串行锁(通过短暂获取 session 锁)
// 同一 topic 的消息处理必须串行执行,防止并发 loop 操作同一历史的不同快照
// 不同 topic 之间互不阻塞,支持多话题并发执行
let serial_lock = {
let mut session_guard = request.session.lock().await;
let lock_key = request.topic_id.as_deref().unwrap_or(request.chat_id);
session_guard.topic_serial_lock(lock_key)
};
// 等待该 topic 的前一条消息处理完成(含压缩)
// await 串行锁时不持有 session 锁,其他 topic 的消息可以正常处理
let _serial_guard = serial_lock.lock().await;
let (history, agent, user_message, user_message_count, original_topic_id) = {
let mut session_guard = request.session.lock().await;
session_guard.ensure_persistent_session(request.chat_id)?;
session_guard.ensure_chat_loaded(request.chat_id)?;
// 优先使用消息接收时捕获的 topic_id消除 #1 与 #2 之间的竞态
let original_topic_id = match &request.topic_id {
Some(tid) => Some(tid.clone()),
None => session_guard
.current_topic(request.chat_id)
.map(|s| s.to_string()),
};
session_guard.ensure_chat_loaded(request.chat_id, original_topic_id.as_deref())?;
session_guard.ensure_agent_prompt_before_user_message(request.chat_id)?;
@ -219,29 +246,29 @@ impl AgentExecutionService {
}
let enriched_content =
enrich_user_content_with_media_refs(request.content, &media_refs)?;
enrich_user_content_with_media_refs(request.content, &media_refs)?;
enrich_user_content_with_media_refs(request.content, &media_refs)?;
// 先计算 user_message_count在添加新消息之前
let history_before = session_guard.get_or_create_history(request.chat_id).clone();
// 无 topic 时用 chat_id 作为 topic_histories 的回退 key
let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id);
let history_before = session_guard.get_or_create_history(history_key).clone();
let user_message_count = history_before.iter().filter(|m| m.role == "user").count();
// 在添加用户消息前,记录当前话题 ID
let original_topic_id = session_guard
.current_topic(request.chat_id)
.map(|s| s.to_string());
let user_message = session_guard.create_user_message(&enriched_content, media_refs);
session_guard.append_persisted_message(request.chat_id, user_message.clone())?;
session_guard.append_persisted_message(
request.chat_id,
original_topic_id.as_deref(),
user_message.clone(),
)?;
// 再获取包含新消息的完整历史记录
let history = session_guard.get_or_create_history(request.chat_id).clone();
let history = session_guard.get_or_create_history(history_key).clone();
session_guard.record_skill_offer(request.chat_id)?;
let mut agent = session_guard.create_agent(
request.chat_id,
Some(request.sender_id),
Some(&user_message.id),
original_topic_id.as_deref(),
)?;
if let Some(handler) = request.live_emitter.clone() {
agent = agent.with_emitted_message_handler(handler);
@ -282,27 +309,46 @@ impl AgentExecutionService {
&self,
request: ScheduledExecutionRequest<'_>,
) -> Result<Vec<OutboundMessage>, AgentError> {
// 获取该 topic 的串行锁(与普通消息路径共享,保证串行执行)
// 定时任务由调度器触发,无用户消息竞态;在锁前一次性捕获 topic_id
// 锁后复用同一值作为 original_topic_id保证锁键与写入目标一致。
let (serial_lock, lock_time_topic_id) = {
let mut session_guard = request.session.lock().await;
let tid = session_guard
.current_topic(request.chat_id)
.map(|s| s.to_string());
let lock_key = tid.as_deref().unwrap_or(request.chat_id);
(session_guard.topic_serial_lock(lock_key), tid)
};
// 等待该 topic 的前一条消息处理完成(含压缩)
let _serial_guard = serial_lock.lock().await;
let (history, mut agent, user_message, user_message_count, original_topic_id, store, session_id) = {
let mut session_guard = request.session.lock().await;
session_guard.ensure_persistent_session(request.chat_id)?;
// 复用锁前捕获的 topic_id保证锁键与写入目标一致
let original_topic_id = lock_time_topic_id.clone();
// 如果 fresh_session 为 true清理历史内存 + 数据库)
if request.fresh_session {
session_guard.clear_chat_history(request.chat_id)?;
session_guard.clear_chat_history(request.chat_id, original_topic_id.as_deref())?;
tracing::info!(
chat_id = %request.chat_id,
"Fresh session enabled, history cleared"
);
}
session_guard.ensure_chat_loaded(request.chat_id)?;
session_guard.ensure_chat_loaded(request.chat_id, original_topic_id.as_deref())?;
session_guard.ensure_agent_prompt_before_user_message(request.chat_id)?;
let scheduled_system_prompt =
compose_scheduled_task_system_prompt(request.system_prompt);
session_guard.append_persisted_message(
request.chat_id,
original_topic_id.as_deref(),
ChatMessage::system_with_context(
&scheduled_system_prompt,
Some(SYSTEM_CONTEXT_SCHEDULED_PROMPT.to_string()),
@ -310,19 +356,19 @@ impl AgentExecutionService {
)?;
// 先计算 user_message_count在添加新消息之前
let history_before = session_guard.get_or_create_history(request.chat_id).clone();
let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id);
let history_before = session_guard.get_or_create_history(history_key).clone();
let user_message_count = history_before.iter().filter(|m| m.role == "user").count();
// 在添加用户消息前,记录当前话题 ID
let original_topic_id = session_guard
.current_topic(request.chat_id)
.map(|s| s.to_string());
let user_message = session_guard.create_user_message(request.prompt, Vec::new());
session_guard.append_persisted_message(request.chat_id, user_message.clone())?;
session_guard.append_persisted_message(
request.chat_id,
original_topic_id.as_deref(),
user_message.clone(),
)?;
// 再获取包含新消息的完整历史记录
let history = session_guard.get_or_create_history(request.chat_id).clone();
let history = session_guard.get_or_create_history(history_key).clone();
session_guard.record_skill_offer(request.chat_id)?;
let agent = session_guard.create_agent_with_provider_config(
@ -331,6 +377,7 @@ impl AgentExecutionService {
Some(request.sender_id),
Some(&user_message.id),
request.provider_config.clone(),
original_topic_id.as_deref(),
)?;
// 获取 store 和 session_id用于构造消息持久化 handler
@ -373,7 +420,7 @@ impl AgentExecutionService {
metadata: request.metadata,
suppress_live_tool_calls: false,
execution_kind: "scheduled_task",
original_topic_id,
original_topic_id: original_topic_id.clone(),
},
)
.await?;
@ -381,7 +428,8 @@ impl AgentExecutionService {
// 清理内存历史,释放内存(数据库历史保留)
{
let mut session_guard = request.session.lock().await;
session_guard.remove_history(request.chat_id);
let history_key = original_topic_id.as_deref().unwrap_or(request.chat_id);
session_guard.remove_history(history_key);
tracing::info!(
chat_id = %request.chat_id,
"Scheduled task completed, memory history released"
@ -399,6 +447,7 @@ impl AgentExecutionService {
let channel_name = request.channel_name.to_string();
let chat_id = request.chat_id.to_string();
let execution_kind = request.execution_kind.to_string();
let topic_id = request.original_topic_id.clone();
let finalized_result = {
let mut session_guard = session.lock().await;
@ -406,8 +455,13 @@ impl AgentExecutionService {
};
if finalized_result.should_schedule_compaction {
if let Err(error) =
schedule_background_history_compaction(session.clone(), chat_id.clone()).await
let compaction_topic_id = topic_id.unwrap_or_else(|| chat_id.clone());
if let Err(error) = schedule_background_history_compaction(
session.clone(),
chat_id.clone(),
compaction_topic_id,
)
.await
{
tracing::warn!(
channel = %channel_name,
@ -481,4 +535,62 @@ mod tests {
assert!(!should_display_message_to_user(false, &message));
assert!(should_display_message_to_user(true, &message));
}
/// 对抗性测试:同一 topic 的串行锁被持有时,第二次获取应阻塞
#[tokio::test]
async fn test_topic_serial_lock_blocks_concurrent_access() {
let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
let _guard1 = lock.lock().await;
// 第二次获取应阻塞1ms 超时验证
let result = tokio::time::timeout(
std::time::Duration::from_millis(1),
lock.lock(),
)
.await;
assert!(result.is_err(), "第二次获取同一锁应阻塞");
}
/// 对抗性测试:不同 topic 的串行锁互不影响,可同时获取
#[tokio::test]
async fn test_different_topic_locks_independent() {
let lock_a = std::sync::Arc::new(tokio::sync::Mutex::new(()));
let lock_b = std::sync::Arc::new(tokio::sync::Mutex::new(()));
let _guard_a = lock_a.lock().await;
// 不同锁应立即可获取
let result = tokio::time::timeout(
std::time::Duration::from_millis(100),
lock_b.lock(),
)
.await;
assert!(result.is_ok(), "不同 topic 的锁应互不影响");
}
/// 对抗性测试错误返回路径锁被正确释放RAII 保证)
#[tokio::test]
async fn test_serial_lock_released_on_error() {
let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
// 模拟 prepare_and_execute_message 的错误路径:
// 获取锁 → 返回错误 → 锁应通过 RAII 释放
{
let _serial_guard = lock.lock().await;
// 模拟错误返回(`?` 或 `Err` 分支)
let _result: Result<(), AgentError> = Err(AgentError::Other("simulated".to_string()));
// _serial_guard 在此块结束时 Drop释放锁
}
// 锁应已释放,可再次获取
let result = tokio::time::timeout(
std::time::Duration::from_millis(100),
lock.lock(),
)
.await;
assert!(result.is_ok(), "错误返回后锁应已释放");
}
}

View File

@ -269,7 +269,7 @@ impl InboundProcessor {
if let Some(ref topic_id) = current_topic {
let cancel_rx = self.cancel_manager.register(topic_id).await;
self.session_manager
.set_agent_cancel_token(&channel, &chat_id, cancel_rx)
.set_agent_cancel_token(&channel, &chat_id, Some(topic_id.as_str()), cancel_rx)
.await;
}
@ -282,6 +282,7 @@ impl InboundProcessor {
&inbound.content,
inbound.media,
Some(live_emitter),
current_topic.as_deref(),
)
.await
{

View File

@ -201,7 +201,6 @@ pub(crate) fn build_session_manager_with_sender(
let runtime_config = SubAgentRuntimeConfig {
default_allowed_tools: task_config.allowed_tools.iter().cloned().collect(),
default_max_execution_secs: task_config.max_execution_secs,
explore_max_execution_secs: task_config.explore_max_execution_secs,
ttl_hours: task_config.ttl_hours,
skills_index: skills.system_index_prompt(),
max_nesting_depth: task_config.max_nesting_depth,

View File

@ -323,14 +323,16 @@ impl Session {
/// 存入待使用的取消信号接收端。
///
/// 在 Agent 执行前由处理器调用Agent 构建时create_agent自动消费。
/// 每个 chat_id 同时只允许一个 pending token新 token 会替换旧 token。
/// 优先按 topic_id 键化(不同 topic 的 token 互不覆盖);
/// 无 topic 时回退到 chat_id。
pub fn set_cancel_receiver(
&mut self,
chat_id: &str,
topic_id: Option<&str>,
receiver: tokio::sync::watch::Receiver<()>,
) {
self.pending_cancel_tokens
.insert(chat_id.to_string(), receiver);
let key = topic_id.unwrap_or(chat_id).to_string();
self.pending_cancel_tokens.insert(key, receiver);
}
/// 获取当前话题 ID指定 chat
@ -338,27 +340,20 @@ impl Session {
self.history.chat_topic(chat_id)
}
/// 获取历史所对应的话题 ID指定 chat
pub fn history_topic(&self, chat_id: &str) -> Option<&str> {
self.history.history_topic(chat_id)
}
/// 切换话题 - 清除当前历史并加载新话题的历史
/// 切换话题 - 设置当前 topic 并加载新话题的历史到内存
/// 不同 topic 的历史在 topic_histories 中独立存储,切换不互斥。
pub fn switch_topic(&mut self, chat_id: &str, topic_id: &str) -> Result<(), AgentError> {
// 清除当前历史
self.history.remove_history(chat_id);
// 先设置当前话题set_history 需要这个)
// 设置当前 topicUI 状态)
self.history.set_chat_topic(chat_id, topic_id.to_string());
// 加载新话题的历史(按 session_id 过滤,排除子智能体消息
// 加载新 topic 的历史到内存(按 topic_id 键化)
let session_id = self.persistent_session_id(chat_id);
let messages = self
.store
.load_messages_for_topic(topic_id, Some(&session_id))
.map_err(|e| AgentError::Other(format!("load topic messages error: {}", e)))?;
self.history.set_history(chat_id, messages);
self.history.set_history(topic_id, messages);
tracing::info!(
topic_id = %topic_id,
@ -372,32 +367,14 @@ impl Session {
self.history.ensure_persistent_session(chat_id)
}
pub fn ensure_chat_loaded(&mut self, chat_id: &str) -> Result<(), AgentError> {
// 检查历史是否存在且对应正确的话题
// 先获取 topic 信息并转换为 owned String避免借用冲突
let current_topic: Option<String> = self.history.chat_topic(chat_id).map(|s| s.to_string());
let stored_topic = self.history.history_topic(chat_id);
if self.chat_history_exists(chat_id) {
// 如果历史已存在,但话题不匹配,需要重新加载
if current_topic.as_deref() != stored_topic {
tracing::info!(
chat_id = %chat_id,
current_topic = ?current_topic,
stored_topic = ?stored_topic,
"Topic changed, reloading history"
);
self.reload_chat_history(chat_id)?;
}
return Ok(());
}
// 历史不存在,按 topic 加载(如果设置了 topic
self.history.ensure_chat_loaded(chat_id, current_topic.as_deref())
}
fn chat_history_exists(&self, chat_id: &str) -> bool {
self.history.get_history(chat_id).is_some()
/// 确保指定 topic 的历史已加载到内存。
/// 按 topic_id 键化查找,已存在则直接返回,否则从 DB 加载。
pub fn ensure_chat_loaded(
&mut self,
chat_id: &str,
topic_id: Option<&str>,
) -> Result<(), AgentError> {
self.history.ensure_chat_loaded(chat_id, topic_id)
}
pub fn ensure_agent_prompt_before_user_message(
@ -408,43 +385,68 @@ impl Session {
.ensure_agent_prompt_before_user_message(chat_id)
}
/// 获取或创建指定 chat_id 的会话历史
pub fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec<ChatMessage> {
self.history.get_or_create_history(chat_id)
/// 获取或创建指定 topic_id 的会话历史
pub fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec<ChatMessage> {
self.history.get_or_create_history(topic_id)
}
/// 获取指定 chat_id 的会话历史(不创建)
pub fn get_history(&self, chat_id: &str) -> Option<&Vec<ChatMessage>> {
self.history.get_history(chat_id)
/// 获取指定 topic_id 的会话历史(不创建)
pub fn get_history(&self, topic_id: &str) -> Option<&Vec<ChatMessage>> {
self.history.get_history(topic_id)
}
/// 使用完整消息追加到历史
pub fn add_message(&mut self, chat_id: &str, message: ChatMessage) {
self.history.add_message(chat_id, message);
/// 使用完整消息追加到指定 topic 的历史
pub fn add_message(&mut self, topic_id: &str, message: ChatMessage) {
self.history.add_message(topic_id, message);
}
pub fn remove_history(&mut self, chat_id: &str) {
self.history.remove_history(chat_id);
pub fn remove_history(&mut self, topic_id: &str) {
self.history.remove_history(topic_id);
}
pub fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
self.history.clear_chat_history(chat_id)
pub fn clear_chat_history(
&mut self,
chat_id: &str,
topic_id: Option<&str>,
) -> Result<(), AgentError> {
self.history.clear_chat_history(chat_id, topic_id)
}
/// 将消息写入内存与持久化层(使用当前 topic
/// 将消息写入内存与持久化层。
/// 优先使用显式传入的 topic_id未传入时回退到当前 chat 的活跃 topic。
/// 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史,
/// 避免旧 topic 的消息污染已切换到的新 topic 的内存历史。
pub fn append_persisted_message(
&mut self,
chat_id: &str,
explicit_topic_id: Option<&str>,
message: ChatMessage,
) -> Result<(), AgentError> {
let session_id = self.persistent_session_id(chat_id);
let topic_id = self.history.chat_topic(chat_id).map(|s| s.to_string());
let topic_id = explicit_topic_id
.map(|s| s.to_string())
.or_else(|| self.history.chat_topic(chat_id).map(|s| s.to_string()));
self.store
.append_message_with_topic(&session_id, topic_id.as_deref(), &message)
.map_err(|err| {
AgentError::Other(format!("append message persistence error: {}", err))
})?;
self.add_message(chat_id, message);
// 只有当写入的 topic 匹配当前活跃 topic 时才更新内存历史。
// 当用户已切换到新 topic 时,旧 topic 的排队消息不应污染新 topic 的内存历史。
let current_chat_topic = self.history.chat_topic(chat_id);
if topic_id.as_deref() == current_chat_topic {
if let Some(ref tid) = topic_id {
self.add_message(tid, message);
}
} else {
tracing::info!(
chat_id = %chat_id,
write_topic_id = ?topic_id,
current_topic_id = ?current_chat_topic,
"Skipping memory history update: message belongs to a different topic"
);
}
// 更新 topic 的最后活跃时间
if let Some(ref topic_id) = topic_id {
@ -458,13 +460,13 @@ impl Session {
pub fn append_persisted_messages<I>(
&mut self,
chat_id: &str,
topic_id: &str,
messages: I,
) -> Result<(), AgentError>
where
I: IntoIterator<Item = ChatMessage>,
{
self.history.append_persisted_messages(chat_id, messages)
self.history.append_persisted_messages(topic_id, messages)
}
/// 将消息保存到指定话题(直接写入数据库,不更新内存历史)
@ -486,32 +488,32 @@ impl Session {
}
#[cfg(test)]
fn latest_user_message_id(&self, chat_id: &str) -> Option<&str> {
self.latest_user_message(chat_id)
fn latest_user_message_id(&self, topic_id: &str) -> Option<&str> {
self.latest_user_message(topic_id)
.map(|message| message.id.as_str())
}
#[cfg(test)]
fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> {
self.history.latest_user_message(chat_id)
fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> {
self.history.latest_user_message(topic_id)
}
#[cfg(test)]
fn is_latest_user_message(&self, chat_id: &str, message_id: &str) -> bool {
self.latest_user_message_id(chat_id)
fn is_latest_user_message(&self, topic_id: &str, message_id: &str) -> bool {
self.latest_user_message_id(topic_id)
.map(|current_id| current_id == message_id)
.unwrap_or(false)
}
pub(crate) fn matches_current_user_turn(&self, chat_id: &str, message: &ChatMessage) -> bool {
self.history.matches_current_user_turn(chat_id, message)
pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool {
self.history.matches_current_user_turn(topic_id, message)
}
pub(crate) fn stale_result_diagnostics(
&self,
chat_id: &str,
topic_id: &str,
) -> (Option<&str>, Option<String>, bool, usize) {
self.history.stale_result_diagnostics(chat_id)
self.history.stale_result_diagnostics(topic_id)
}
/// 清除所有历史
@ -533,20 +535,20 @@ impl Session {
&self.compressor
}
pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
// 如果当前有 topic加载该 topic 的消息(按 session_id 过滤,排除子智能体消息)
if let Some(topic_id) = self.history.chat_topic(chat_id) {
let session_id = self.persistent_session_id(chat_id);
let messages = self
.store
.load_messages_for_topic(topic_id, Some(&session_id))
.map_err(|e| AgentError::Other(format!("load topic messages error: {}", e)))?;
self.history.set_history(chat_id, messages);
} else {
// 否则加载 session 的所有消息
self.history.reload_chat_history(chat_id)?;
}
Ok(())
/// 获取该 topic 的串行化锁。
/// 同一 topic 的消息处理agent loop + 压缩)共享此锁,保证串行执行;
/// 不同 topic 之间互不阻塞。
pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc<tokio::sync::Mutex<()>> {
self.history.topic_serial_lock(topic_id)
}
/// 按 topic_id 从 DB 重新加载历史到内存
pub(crate) fn reload_topic_history(
&mut self,
chat_id: &str,
topic_id: &str,
) -> Result<(), AgentError> {
self.history.reload_topic_history(chat_id, topic_id)
}
pub(crate) fn store(&self) -> Arc<dyn ConversationRepository> {
@ -572,6 +574,7 @@ impl Session {
chat_id: &str,
sender_id: Option<&str>,
message_id: Option<&str>,
explicit_topic_id: Option<&str>,
) -> Result<AgentLoop, AgentError> {
self.create_agent_with_provider_config(
chat_id,
@ -579,6 +582,7 @@ impl Session {
sender_id,
message_id,
self.provider_config.clone(),
explicit_topic_id,
)
}
@ -589,10 +593,21 @@ impl Session {
sender_id: Option<&str>,
message_id: Option<&str>,
provider_config: LLMProviderConfig,
explicit_topic_id: Option<&str>,
) -> Result<AgentLoop, AgentError> {
// 优先使用显式传入的 topic_id回退到当前 chat 的活跃 topic
let topic_id = explicit_topic_id
.map(|s| s.to_string())
.or_else(|| self.current_topic(session_chat_id).map(|s| s.to_string()));
// 消费 pending 的取消信号接收端(如果存在)
let cancel_token = self.pending_cancel_tokens.remove(session_chat_id);
let topic_id = self.current_topic(session_chat_id).map(|s| s.to_string());
// 优先按 topic_id 查找;无 topic 时回退 chat_id
let cancel_token = match &topic_id {
Some(tid) => self.pending_cancel_tokens.remove(tid)
.or_else(|| self.pending_cancel_tokens.remove(session_chat_id)),
None => self.pending_cancel_tokens.remove(session_chat_id),
};
self.agent_factory.create(AgentBuildRequest {
channel_name: &self.channel_name,
session_chat_id,
@ -812,10 +827,11 @@ impl SessionManager {
&self,
channel_name: &str,
chat_id: &str,
topic_id: Option<&str>,
token: tokio::sync::watch::Receiver<()>,
) {
if let Some(session) = self.get(channel_name).await {
session.lock().await.set_cancel_receiver(chat_id, token);
session.lock().await.set_cancel_receiver(chat_id, topic_id, token);
}
}
@ -837,6 +853,7 @@ impl SessionManager {
content: &str,
media: Vec<crate::bus::MediaItem>,
live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
topic_id: Option<&str>,
) -> Result<Vec<OutboundMessage>, AgentError> {
self.messages
.handle_message(
@ -846,6 +863,7 @@ impl SessionManager {
content,
media,
live_emitter,
topic_id,
)
.await
}
@ -953,7 +971,7 @@ mod tests {
user_tx,
tools,
skills,
store,
store.clone(),
100,
Arc::new(SubagentRuntime::from_config(Default::default())),
)
@ -961,19 +979,22 @@ mod tests {
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
let session_id = session.persistent_session_id("chat-1");
let topic = store.create_topic(&session_id, "test topic", None).unwrap();
let topic_id = topic.id.clone();
session.switch_topic("chat-1", &topic_id).unwrap();
let first = session.create_user_message("first", Vec::new());
let first_id = first.id.clone();
session.append_persisted_message("chat-1", first).unwrap();
assert!(session.is_latest_user_message("chat-1", &first_id));
session.append_persisted_message("chat-1", Some(&topic_id), first).unwrap();
assert!(session.is_latest_user_message(&topic_id, &first_id));
let second = session.create_user_message("second", Vec::new());
let second_id = second.id.clone();
session.append_persisted_message("chat-1", second).unwrap();
session.append_persisted_message("chat-1", Some(&topic_id), second).unwrap();
assert!(!session.is_latest_user_message("chat-1", &first_id));
assert!(session.is_latest_user_message("chat-1", &second_id));
assert!(!session.is_latest_user_message(&topic_id, &first_id));
assert!(session.is_latest_user_message(&topic_id, &second_id));
}
#[tokio::test]
@ -1010,46 +1031,37 @@ mod tests {
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
let session_id = session.persistent_session_id("chat-1");
let topic = store.create_topic(&session_id, "test topic", None).unwrap();
let topic_id = topic.id.clone();
session.switch_topic("chat-1", &topic_id).unwrap();
let first = session.create_user_message("first", Vec::new());
let first_id = first.id.clone();
session.append_persisted_message("chat-1", first).unwrap();
session.append_persisted_message("chat-1", Some(&topic_id), first).unwrap();
session
.append_persisted_message("chat-1", ChatMessage::assistant("answer-1"))
.append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-1"))
.unwrap();
let second = session.create_user_message("second", Vec::new());
session
.append_persisted_message("chat-1", second.clone())
.append_persisted_message("chat-1", Some(&topic_id), second.clone())
.unwrap();
session
.append_persisted_message("chat-1", ChatMessage::assistant("answer-2"))
.append_persisted_message("chat-1", Some(&topic_id), ChatMessage::assistant("answer-2"))
.unwrap();
let session_id = session.persistent_session_id("chat-1");
let snapshot_end_seq = store
.get_session(&session_id)
.unwrap()
.unwrap()
.message_count;
let preserved_messages = session.get_history("chat-1").unwrap().clone();
let preserved_messages = session.get_history(&topic_id).unwrap().clone();
store
.compact_active_history(
&session_id,
snapshot_end_seq,
&[],
&ChatMessage::system("[Compressed History]\n\nsummary"),
&preserved_messages,
)
.replace_topic_history(&session_id, &topic_id, &preserved_messages)
.unwrap();
session.reload_chat_history("chat-1").unwrap();
session.reload_topic_history("chat-1", &topic_id).unwrap();
assert!(!session.is_latest_user_message("chat-1", &first_id));
assert!(!session.is_latest_user_message("chat-1", &second.id));
assert!(session.matches_current_user_turn("chat-1", &second));
assert!(!session.is_latest_user_message(&topic_id, &first_id));
assert!(session.is_latest_user_message(&topic_id, &second.id));
assert!(session.matches_current_user_turn(&topic_id, &second));
}
async fn start_mock_openai_server() -> String {
@ -1209,7 +1221,7 @@ mod tests {
.unwrap();
let outbound = session_manager
.handle_message("test-channel", "user-1", "chat-1", "hello", Vec::new(), None)
.handle_message("test-channel", "user-1", "chat-1", "hello", Vec::new(), None, None)
.await
.unwrap();
@ -2037,7 +2049,7 @@ mod tests {
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
session.switch_topic("chat-1", "chat-1").unwrap();
let history = session.get_history("chat-1").unwrap();
// 新设计:系统提示词不再持久化到历史记录,而是每次请求时动态注入
@ -2078,11 +2090,14 @@ mod tests {
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
let session_id = session.persistent_session_id("chat-1");
let topic = store.create_topic(&session_id, "test topic", None).unwrap();
let topic_id = topic.id.clone();
session.switch_topic("chat-1", &topic_id).unwrap();
for turn in 0..100 {
session
.append_persisted_message("chat-1", ChatMessage::user(format!("user-{turn}")))
.append_persisted_message("chat-1", Some(&topic_id), ChatMessage::user(format!("user-{turn}")))
.unwrap();
}
@ -2091,7 +2106,7 @@ mod tests {
.unwrap();
// 新设计:系统提示词不再持久化到历史记录
let history = session.get_history("chat-1").unwrap();
let history = session.get_history(&topic_id).unwrap();
let user_messages = history
.iter()
.filter(|message| message.role == "user")
@ -2110,7 +2125,7 @@ mod tests {
session
.ensure_agent_prompt_before_user_message("chat-1")
.unwrap();
let history = session.get_history("chat-1").unwrap();
let history = session.get_history(&topic_id).unwrap();
let user_messages = history
.iter()
.filter(|message| message.role == "user")
@ -2152,11 +2167,14 @@ mod tests {
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
let session_id = session.persistent_session_id("chat-1");
let topic = store.create_topic(&session_id, "test topic", None).unwrap();
let topic_id = topic.id.clone();
session.switch_topic("chat-1", &topic_id).unwrap();
for turn in 0..100 {
session
.append_persisted_message("chat-1", ChatMessage::user(format!("user-{turn}")))
.append_persisted_message("chat-1", Some(&topic_id), ChatMessage::user(format!("user-{turn}")))
.unwrap();
}
@ -2165,7 +2183,7 @@ mod tests {
.unwrap();
// 新设计:系统提示词不再持久化到历史记录
let history = session.get_history("chat-1").unwrap();
let history = session.get_history(&topic_id).unwrap();
let user_messages = history
.iter()
.filter(|message| message.role == "user")

View File

@ -17,10 +17,18 @@ fn preview_text(content: &str, max_chars: usize) -> String {
pub(crate) struct SessionHistory {
channel_name: String,
chat_histories: HashMap<String, Vec<ChatMessage>>,
chat_topic_ids: HashMap<String, String>, // 每个 chat 的当前 topic
history_topic_ids: HashMap<String, String>, // 每个 chat 的历史所对应的话题
/// 按 topic_id 键化的内存历史缓存。
/// 不同 topic 的历史独立存储,互不干扰,支持多话题并发执行。
topic_histories: HashMap<String, Vec<ChatMessage>>,
/// UI 状态:每个 chat 当前活跃的 topic按 chat_id 键)。
chat_topic_ids: HashMap<String, String>,
/// 正在压缩中的 topic_id 集合
compression_in_flight: HashSet<String>,
/// 按 topic_id 的串行化锁。
/// 同一 topic 的消息处理agent loop + 压缩)必须串行执行,
/// 防止并发 loop 操作同一历史的不同快照产生交错序列。
/// 不同 topic 之间互不阻塞,支持多话题并发执行。
topic_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
conversations: Arc<dyn ConversationRepository>,
skill_events: Arc<dyn SkillEventRepository>,
}
@ -33,15 +41,25 @@ impl SessionHistory {
) -> Self {
Self {
channel_name: channel_name.into(),
chat_histories: HashMap::new(),
topic_histories: HashMap::new(),
chat_topic_ids: HashMap::new(),
history_topic_ids: HashMap::new(),
compression_in_flight: HashSet::new(),
topic_serial_locks: HashMap::new(),
conversations,
skill_events,
}
}
/// 获取或创建该 topic 的串行化锁。
/// 同一 topic 的所有消息处理共享同一个锁,保证串行执行;
/// 不同 topic 之间互不阻塞,支持多话题并发执行。
pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc<tokio::sync::Mutex<()>> {
self.topic_serial_locks
.entry(topic_id.to_string())
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
.clone()
}
pub(crate) fn persistent_session_id(&self, chat_id: &str) -> String {
persistent_session_id(&self.channel_name, chat_id)
}
@ -55,38 +73,36 @@ impl SessionHistory {
.map_err(|err| AgentError::Other(format!("session persistence error: {}", err)))
}
/// 确保指定 topic 的历史已加载到内存。
/// 按 topic_id 键化查找,如果已存在则直接返回,否则从 DB 加载。
pub(crate) fn ensure_chat_loaded(
&mut self,
chat_id: &str,
topic_id: Option<&str>,
) -> Result<(), AgentError> {
if self.chat_histories.contains_key(chat_id) {
let Some(tid) = topic_id else {
return Ok(());
};
if self.topic_histories.contains_key(tid) {
return Ok(());
}
// 如果提供了 topic_id按 topic 加载;否则按 session 加载
let mut history = if let Some(tid) = topic_id {
let sid = self.persistent_session_id(chat_id);
self.conversations
.load_messages_for_topic(tid, Some(&sid))
.map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?
} else {
self.conversations
.load_messages(&self.persistent_session_id(chat_id))
.map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?
};
let sid = self.persistent_session_id(chat_id);
let mut history = self
.conversations
.load_messages_for_topic(tid, Some(&sid))
.map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?;
// 清理 DB 加载的历史中可能存在的不完整 tool_call 序列
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history);
if removed > 0 {
tracing::warn!(
chat_id = %chat_id,
topic_id = %tid,
removed_count = removed,
"Sanitized incomplete tool_call sequences on history load"
);
}
self.chat_histories.insert(chat_id.to_string(), history);
self.topic_histories.insert(tid.to_string(), history);
Ok(())
}
@ -98,58 +114,56 @@ impl SessionHistory {
Ok(())
}
pub(crate) fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec<ChatMessage> {
self.chat_histories.entry(chat_id.to_string()).or_default()
pub(crate) fn get_or_create_history(&mut self, topic_id: &str) -> &mut Vec<ChatMessage> {
self.topic_histories.entry(topic_id.to_string()).or_default()
}
pub(crate) fn get_history(&self, chat_id: &str) -> Option<&Vec<ChatMessage>> {
self.chat_histories.get(chat_id)
pub(crate) fn get_history(&self, topic_id: &str) -> Option<&Vec<ChatMessage>> {
self.topic_histories.get(topic_id)
}
pub(crate) fn set_history(&mut self, chat_id: &str, history: Vec<ChatMessage>) {
self.chat_histories.insert(chat_id.to_string(), history);
// 记录历史对应的话题(当前设置的话题)
if let Some(topic_id) = self.chat_topic_ids.get(chat_id) {
self.history_topic_ids.insert(chat_id.to_string(), topic_id.clone());
}
pub(crate) fn set_history(&mut self, topic_id: &str, history: Vec<ChatMessage>) {
self.topic_histories.insert(topic_id.to_string(), history);
}
/// 获取指定 chat 的历史所对应的话题
pub(crate) fn history_topic(&self, chat_id: &str) -> Option<&str> {
self.history_topic_ids.get(chat_id).map(|s| s.as_str())
}
/// 设置指定 chat 的当前 topic
/// 设置指定 chat 的当前 topicUI 状态)
pub(crate) fn set_chat_topic(&mut self, chat_id: &str, topic_id: String) {
self.chat_topic_ids.insert(chat_id.to_string(), topic_id);
}
/// 获取指定 chat 的当前 topic
/// 获取指定 chat 的当前 topicUI 状态)
pub(crate) fn chat_topic(&self, chat_id: &str) -> Option<&str> {
self.chat_topic_ids.get(chat_id).map(|s| s.as_str())
}
/// 清除指定 chat 的 topic
/// 清除指定 chat 的 topicUI 状态)
pub(crate) fn clear_chat_topic(&mut self, chat_id: &str) {
self.chat_topic_ids.remove(chat_id);
}
pub(crate) fn add_message(&mut self, chat_id: &str, message: ChatMessage) {
self.get_or_create_history(chat_id).push(message);
pub(crate) fn add_message(&mut self, topic_id: &str, message: ChatMessage) {
self.get_or_create_history(topic_id).push(message);
}
pub(crate) fn remove_history(&mut self, chat_id: &str) {
self.chat_histories.remove(chat_id);
self.compression_in_flight.remove(chat_id);
self.history_topic_ids.remove(chat_id);
pub(crate) fn remove_history(&mut self, topic_id: &str) {
self.topic_histories.remove(topic_id);
self.compression_in_flight.remove(topic_id);
}
pub(crate) fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
if let Some(history) = self.chat_histories.get_mut(chat_id) {
let len = history.len();
history.clear();
#[cfg(debug_assertions)]
tracing::debug!(chat_id = %chat_id, previous_len = len, "Chat history cleared");
/// 清空指定 chat/topic 的内存历史和 DB 消息。
/// 内存按 topic_id 清DB 按 session_id 清(保留原行为以兼容无 topic 场景)。
pub(crate) fn clear_chat_history(
&mut self,
chat_id: &str,
topic_id: Option<&str>,
) -> Result<(), AgentError> {
if let Some(tid) = topic_id {
if let Some(history) = self.topic_histories.get_mut(tid) {
let len = history.len();
history.clear();
#[cfg(debug_assertions)]
tracing::debug!(topic_id = %tid, previous_len = len, "Topic history cleared");
}
}
self.conversations
@ -159,7 +173,7 @@ impl SessionHistory {
pub(crate) fn append_persisted_messages<I>(
&mut self,
chat_id: &str,
topic_id: &str,
messages: I,
) -> Result<(), AgentError>
where
@ -170,13 +184,11 @@ impl SessionHistory {
return Ok(());
}
// 在追加新消息前,先清理内存历史中的不完整 tool_call 序列
// 这防止脏数据(如取消时产生的孤立 assistant(tool_calls))在内存历史中累积
if let Some(history) = self.chat_histories.get_mut(chat_id) {
if let Some(history) = self.topic_histories.get_mut(topic_id) {
let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(history);
if removed > 0 {
tracing::warn!(
chat_id = %chat_id,
topic_id = %topic_id,
removed_count = removed,
"Sanitized in-memory history before appending persisted messages"
);
@ -184,7 +196,7 @@ impl SessionHistory {
}
for message in messages {
self.add_message(chat_id, message);
self.add_message(topic_id, message);
}
Ok(())
}
@ -203,13 +215,13 @@ impl SessionHistory {
Ok(())
}
pub(crate) fn latest_user_message(&self, chat_id: &str) -> Option<&ChatMessage> {
self.get_history(chat_id)
pub(crate) fn latest_user_message(&self, topic_id: &str) -> Option<&ChatMessage> {
self.get_history(topic_id)
.and_then(|history| history.iter().rev().find(|message| message.role == "user"))
}
pub(crate) fn matches_current_user_turn(&self, chat_id: &str, message: &ChatMessage) -> bool {
self.latest_user_message(chat_id)
pub(crate) fn matches_current_user_turn(&self, topic_id: &str, message: &ChatMessage) -> bool {
self.latest_user_message(topic_id)
.map(|current| {
current.id == message.id
|| (current.content == message.content
@ -221,14 +233,14 @@ impl SessionHistory {
pub(crate) fn stale_result_diagnostics(
&self,
chat_id: &str,
topic_id: &str,
) -> (Option<&str>, Option<String>, bool, usize) {
let latest_user = self.latest_user_message(chat_id);
let latest_user = self.latest_user_message(topic_id);
let latest_user_id = latest_user.map(|message| message.id.as_str());
let latest_user_preview = latest_user.map(|message| preview_text(&message.content, 80));
let compression_in_flight = self.compression_in_flight.contains(chat_id);
let compression_in_flight = self.compression_in_flight.contains(topic_id);
let history_len = self
.get_history(chat_id)
.get_history(topic_id)
.map(|history| history.len())
.unwrap_or(0);
@ -240,31 +252,29 @@ impl SessionHistory {
)
}
/// 清空所有内存历史(主要用于测试全局重置)。
/// 不遍历清 DB生产环境如需清 DB 应由调用方显式调用。
pub(crate) fn clear_all_history(&mut self) -> Result<(), AgentError> {
let chat_ids: Vec<String> = self.chat_histories.keys().cloned().collect();
let total: usize = self.chat_histories.values().map(|h| h.len()).sum();
self.chat_histories.clear();
let total: usize = self.topic_histories.values().map(|h| h.len()).sum();
self.topic_histories.clear();
self.compression_in_flight.clear();
#[cfg(debug_assertions)]
tracing::debug!(previous_total = total, "All chat histories cleared");
for chat_id in chat_ids {
self.conversations
.clear_messages(&self.persistent_session_id(&chat_id))
.map_err(|err| {
AgentError::Other(format!("clear history persistence error: {}", err))
})?;
}
tracing::debug!(previous_total = total, "All topic histories cleared");
Ok(())
}
pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
/// 按 topic_id 从 DB 重新加载历史到内存
pub(crate) fn reload_topic_history(
&mut self,
chat_id: &str,
topic_id: &str,
) -> Result<(), AgentError> {
let sid = self.persistent_session_id(chat_id);
let history = self
.conversations
.load_messages(&self.persistent_session_id(chat_id))
.load_messages_for_topic(topic_id, Some(&sid))
.map_err(|err| AgentError::Other(format!("session history reload error: {}", err)))?;
self.chat_histories.insert(chat_id.to_string(), history);
self.topic_histories.insert(topic_id.to_string(), history);
Ok(())
}

View File

@ -28,6 +28,7 @@ impl SessionMessageService {
content: &str,
media: Vec<MediaItem>,
live_emitter: Option<Arc<dyn EmittedMessageHandler>>,
topic_id: Option<&str>,
) -> Result<Vec<OutboundMessage>, AgentError> {
#[cfg(debug_assertions)]
{
@ -54,6 +55,7 @@ impl SessionMessageService {
content,
media,
live_emitter,
topic_id: topic_id.map(|s| s.to_string()),
})
.await?;

View File

@ -36,152 +36,49 @@ impl SystemPromptProvider for ToolPromptProvider {
/// memory_search / memory_manage 工具使用说明
const MEMORY_TOOLS_INSTRUCTIONS: &str = r#"# 记忆工具
##
使 memory_search
###
- 使 memory_search memory_search(action='search')
- namespace key get
- list
- 使
###
-
-
###
- queries 10-12
##
- queries 10-12
-
-
- queries=['email', '', 'folder',"preference"]
-
##
###
使
- `user` -
- `semantic` -
- `episodic` -
- `skill` -
- `environment` -
- `reflection` -
- `other` -
###
- 使 memory_manage
-
###
-
##
使 memory_manage
-
-
- //
- /
-
- xxx的消息
-
- //
- /
-
- "默认 xxx"
###
- 使 memory_manage "#;
##
- memory_manage "#;
/// skill_activate / skill_manage 工具使用说明
const SKILL_TOOLS_INSTRUCTIONS: &str = r#"# 技能工具
##
- : `{project-root}/.picobot/skills/{skill-name}/SKILL.md`
- : `~/.picobot/skills/{skill-name}/SKILL.md`
## /
- 使 `skill_manage` `create` `update` action
- 使 `write`
- `skill_manage`
## 使
## 使
-
- `skill_activate`
-
- `skill_activate`
-
- / `skill_manage` `create` / `update` action `write`
## 使
使
-
## 使
-
-
-
## 使
1. ****: <available_skills>
2. ****:
3. ****: `skill_activate` name
4. ****: skill_activate "#;
- "#;
/// todo_write / todo_read 工具使用说明
const TODO_WRITE_INSTRUCTIONS: &str = r#"# TodoWrite 工具
使 `todo_write`
## 使
- 3 使 todo_write
- todo
- todo
- todo
-
-
## merge
- `merge: true` **使**
- `merge: false` todo
##
- `pending`
- `in_progress`
- `completed`
- `cancelled`
##
1. `in_progress`
2. `in_progress`
3. `completed` `cancelled` `in_progress` `pending`
4. `in_progress` 退 `pending` `completed` `cancelled`
5. completed
6. `content`
7. ** `id`** id `"r9Tg8Kq2"`使 idid todo_write `current_todos`
## 使
id
```json
{"merge": true, "todos": [{"id": "aB3kLm9x", "content": "修复登录 bug", "status": "in_progress"}]}
```
```json
{"merge": true, "todos": [{"id": "pQ7nWy2z", "content": "补充测试", "status": "pending"}]}
```
使 id
```json
{"merge": true, "todos": [{"id": "aB3kLm9x", "content": "修复登录 bug", "status": "completed"}]}
```
```json
{"merge": true, "todos": [
{"id": "aB3kLm9x", "content": "修复登录 bug", "status": "completed"},
{"id": "pQ7nWy2z", "content": "补充测试", "status": "in_progress"}
]}
```
##
- todo
-
-
- `in_progress` 退 `pending` `completed` `cancelled`
- `completed` / `cancelled`
##
使 `todo_read`
```json
{}
```
`todo_read`
-
-
- "#;
- `todo_read` "#;
/// shell / bash 工具使用说明
const SHELL_TOOLS_INSTRUCTIONS: &str = r#"# Shell 交互终端

View File

@ -8,6 +8,7 @@ pub mod command;
pub mod config;
pub mod domain;
pub mod experts;
pub mod frontmatter;
pub mod gateway;
pub mod logging;
pub mod mcp;

View File

@ -88,6 +88,9 @@ pub struct TodoItemSummary {
pub id: String,
pub content: String,
pub status: String,
pub priority: String,
pub created_at: i64,
pub updated_at: i64,
pub created_by_message_id: Option<String>,
}

View File

@ -394,6 +394,17 @@ impl OpenAIProvider {
let status = resp.status();
if !status.is_success() {
let text = resp.text().await.unwrap_or_default();
let sequence = format_message_sequence(&body);
tracing::error!(
provider = %self.name,
model = %self.model_id,
url = %url,
status = %status,
response_len = text.len(),
response_body = %text,
sequence = ?sequence,
"OpenAI-compatible streaming API request failed"
);
return Err(format!("API error {}: {}", status, text).into());
}
@ -672,6 +683,82 @@ impl OpenAIProvider {
}
}
// Forward-order check: verify tool messages IMMEDIATELY follow the
// assistant(tool_calls). If any non-tool message appears between the
// assistant and its tool results, the API rejects with
// "insufficient tool messages following tool_calls message".
//
// The reverse scan above only checks existence (tool result appears
// somewhere after assistant), NOT immediacy. This forward pass catches:
// [assistant(tool_calls=[A]), user, tool(A)]
// ^ reverse scan sees tool(A) after assistant → "resolved"
// but API requires tool(A) to be IMMEDIATELY after assistant
{
let mut pending_tool_ids: std::collections::HashSet<&str> = std::collections::HashSet::new();
let mut pending_assistant_idx: Option<usize> = None;
for (i, m) in request.messages.iter().enumerate() {
// If we have pending tool_ids and encounter a non-tool message,
// the assistant's tool results were NOT immediately following.
if !pending_tool_ids.is_empty() && m.role != "tool" {
if let Some(idx) = pending_assistant_idx {
skip_assistant_indices.insert(idx);
tracing::warn!(
message_index = idx,
interrupted_by_index = i,
interrupted_by_role = %m.role,
pending_tool_call_count = pending_tool_ids.len(),
"build_request_body: assistant tool_calls not immediately \
followed by tool results stripping tool_calls"
);
// Remove this assistant's tool_call_ids from with_parent
// so orphaned tool messages are dropped during serialization
if let Some(calls) = &request.messages[idx].tool_calls {
for tc in calls.iter() {
with_parent.remove(tc.id.as_str());
}
}
}
pending_tool_ids.clear();
pending_assistant_idx = None;
}
if m.role == "assistant" {
if let Some(ref calls) = m.tool_calls {
if !calls.is_empty() && !skip_assistant_indices.contains(&i) {
pending_tool_ids = calls.iter().map(|tc| tc.id.as_str()).collect();
pending_assistant_idx = Some(i);
}
}
} else if m.role == "tool" {
if let Some(ref tc_id) = m.tool_call_id {
pending_tool_ids.remove(tc_id.as_str());
if pending_tool_ids.is_empty() {
pending_assistant_idx = None;
}
}
}
}
// Handle trailing assistant with unresolved immediate tool results
if !pending_tool_ids.is_empty() {
if let Some(idx) = pending_assistant_idx {
skip_assistant_indices.insert(idx);
tracing::warn!(
message_index = idx,
pending_tool_call_count = pending_tool_ids.len(),
"build_request_body: trailing assistant tool_calls without \
immediately following tool results stripping tool_calls"
);
if let Some(calls) = &request.messages[idx].tool_calls {
for tc in calls.iter() {
with_parent.remove(tc.id.as_str());
}
}
}
}
}
// valid_tool_call_parent_ids = with_parent (assistant tool_call_ids
// whose parent assistant has ALL results after it)
let valid_tool_call_parent_ids = &with_parent;
@ -799,6 +886,38 @@ impl OpenAIProvider {
}
}
/// Builds a compact, human-readable summary of the message sequence in `body`
/// for diagnostic logging. Only emitted on API errors (e.g. 400 responses) to
/// avoid flooding logs on every request — see callers in `chat` and
/// `chat_streaming_internal`.
fn format_message_sequence(body: &Value) -> Vec<String> {
body["messages"].as_array()
.map(|msgs| msgs.iter().enumerate().map(|(i, m)| {
let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("?");
match role {
"assistant" => {
let tc_count = m.get("tool_calls")
.and_then(|t| t.as_array())
.map(|a| a.len())
.unwrap_or(0);
if tc_count > 0 {
format!("[{}] assistant(tool_calls={})", i, tc_count)
} else {
format!("[{}] assistant", i)
}
}
"tool" => {
let tcid = m.get("tool_call_id")
.and_then(|t| t.as_str())
.unwrap_or("??");
format!("[{}] tool(id={})", i, tcid)
}
_ => format!("[{}] {}", i, role),
}
}).collect())
.unwrap_or_default()
}
#[derive(Deserialize)]
struct OpenAIResponse {
id: String,
@ -945,6 +1064,7 @@ impl LLMProvider for OpenAIProvider {
// Debug: Log LLM response (only in debug builds)
if !status.is_success() {
let sequence = format_message_sequence(&body);
tracing::error!(
provider = %self.name,
model = %self.model_id,
@ -952,6 +1072,7 @@ impl LLMProvider for OpenAIProvider {
status = %status,
response_len = text.len(),
response_body = %text,
sequence = ?sequence,
"OpenAI-compatible API request failed"
);
return Err(format!("API error {}: {}", status, text).into());
@ -1496,4 +1617,131 @@ mod tests {
// custom_param 应该保留
assert_eq!(body["custom_param"], Value::String("value".to_string()));
}
#[test]
fn test_build_request_body_strips_tool_calls_when_not_immediately_followed() {
// [assistant(tool_calls=[A]), user, tool(A)] → should strip tool_calls
// The tool result exists but a user message interrupts between assistant
// and tool result. The API would reject this with
// "insufficient tool messages following tool_calls message".
let provider = OpenAIProvider::new(
"test".to_string(),
"key".to_string(),
"https://example.com/v1".to_string(),
HashMap::new(),
120,
"gpt-test".to_string(),
None,
None,
HashMap::new(),
);
let request = ChatCompletionRequest {
messages: vec![
Message {
role: "assistant".to_string(),
content: vec![ContentBlock::text("calling tool")],
reasoning_content: None,
tool_call_id: None,
name: None,
tool_calls: Some(vec![ToolCall {
id: "call_A".to_string(),
name: "search".to_string(),
arguments: json!({}),
}]),
},
Message {
role: "user".to_string(),
content: vec![ContentBlock::text("interrupting message")],
reasoning_content: None,
tool_call_id: None,
name: None,
tool_calls: None,
},
Message {
role: "tool".to_string(),
content: vec![ContentBlock::text("result")],
reasoning_content: None,
tool_call_id: Some("call_A".to_string()),
name: Some("search".to_string()),
tool_calls: None,
},
],
temperature: None,
max_tokens: None,
tools: None,
};
let body = provider.build_request_body(&request);
let messages = body["messages"].as_array().unwrap();
// Assistant should NOT have tool_calls (stripped because not immediately followed)
assert!(
messages[0].get("tool_calls").is_none(),
"tool_calls should be stripped when tool results are not immediately following"
);
// Tool message should be dropped (orphaned after stripping)
assert_eq!(
messages.len(),
2,
"tool message should be dropped as orphaned, got {} messages",
messages.len()
);
}
#[test]
fn test_build_request_body_preserves_tool_calls_when_immediately_followed() {
// [assistant(tool_calls=[A]), tool(A)] → should keep tool_calls (valid sequence)
let provider = OpenAIProvider::new(
"test".to_string(),
"key".to_string(),
"https://example.com/v1".to_string(),
HashMap::new(),
120,
"gpt-test".to_string(),
None,
None,
HashMap::new(),
);
let request = ChatCompletionRequest {
messages: vec![
Message {
role: "assistant".to_string(),
content: vec![ContentBlock::text("calling tool")],
reasoning_content: None,
tool_call_id: None,
name: None,
tool_calls: Some(vec![ToolCall {
id: "call_A".to_string(),
name: "search".to_string(),
arguments: json!({}),
}]),
},
Message {
role: "tool".to_string(),
content: vec![ContentBlock::text("result")],
reasoning_content: None,
tool_call_id: Some("call_A".to_string()),
name: Some("search".to_string()),
tool_calls: None,
},
],
temperature: None,
max_tokens: None,
tools: None,
};
let body = provider.build_request_body(&request);
let messages = body["messages"].as_array().unwrap();
// Assistant should keep tool_calls (valid immediate sequence)
let tool_calls = messages[0].get("tool_calls")
.and_then(|t| t.as_array())
.expect("tool_calls should be preserved when immediately followed");
assert_eq!(tool_calls.len(), 1);
// Tool message should be present
assert_eq!(messages.len(), 2);
assert_eq!(messages[1]["role"], "tool");
}
}

View File

@ -867,11 +867,14 @@ struct SkillFrontmatter {
fn parse_skill_file(path: &Path, source: SkillSource) -> Result<Skill, String> {
let content = fs::read_to_string(path).map_err(|e| format!("failed to read file: {}", e))?;
let (frontmatter_raw, body) =
split_frontmatter(&content).ok_or_else(|| "missing YAML frontmatter block".to_string())?;
let frontmatter: SkillFrontmatter = serde_yaml::from_str(frontmatter_raw)
.map_err(|e| format!("invalid YAML frontmatter: {}", e))?;
let (frontmatter, body) = match crate::frontmatter::parse::<SkillFrontmatter>(&content) {
Ok(v) => v,
Err(err) => {
let bytes = content.len();
let crlf = content.contains('\r');
return Err(format!("{} (bytes={}, crlf={})", err, bytes, crlf));
}
};
let description = frontmatter.description.trim();
if description.is_empty() {
@ -895,15 +898,6 @@ fn parse_skill_file(path: &Path, source: SkillSource) -> Result<Skill, String> {
})
}
fn split_frontmatter(content: &str) -> Option<(&str, &str)> {
let rest = content.strip_prefix("---\n")?;
let marker = "\n---\n";
let idx = rest.find(marker)?;
let frontmatter = &rest[..idx];
let body = &rest[idx + marker.len()..];
Some((frontmatter, body))
}
// 使用 platform 模块提供的 xml_escape 和 path_to_uri 函数
// SkillPromptProvider 实现
@ -999,11 +993,17 @@ mod tests {
}
#[test]
fn test_split_frontmatter() {
let input = "---\ndescription: demo\n---\nhello";
let (fm, body) = split_frontmatter(input).unwrap();
assert!(fm.contains("description"));
assert_eq!(body, "hello");
fn test_parse_skill_file_handles_crlf_endings() {
let dir = tempfile::tempdir().unwrap();
let skill_dir = dir.path().join("demo");
fs::create_dir_all(&skill_dir).unwrap();
let skill_md = skill_dir.join("SKILL.md");
fs::write(&skill_md, "---\r\ndescription: demo skill\r\n---\r\nStep A\r\nStep B").unwrap();
let skill = parse_skill_file(&skill_md, SkillSource::Project).unwrap();
assert_eq!(skill.name, "demo");
assert_eq!(skill.description, "demo skill");
assert_eq!(skill.body, "Step A\nStep B");
}
#[test]

View File

@ -875,6 +875,74 @@ impl SessionStore {
Ok(())
}
/// Replace the entire history for a specific topic.
///
/// Deletes only messages belonging to the given topic_id (preserving
/// other topics' messages), then inserts the new messages with topic_id
/// set correctly. Used by the compressor when it has produced a
/// complete, validated message list for a single topic.
///
/// Seq numbers continue from the current session-wide max (not reset to
/// 1) so we don't collide with other topics' messages. Gaps in seq
/// (from the deleted old messages) are harmless — per-topic loading
/// orders by seq and gaps don't affect ordering.
pub fn replace_topic_history(
&self,
session_id: &str,
topic_id: &str,
messages: &[ChatMessage],
) -> Result<(), StorageError> {
let conn = self.pool.get()?;
let tx = conn.unchecked_transaction()?;
let now = current_timestamp();
// Delete only messages belonging to this topic — other topics'
// messages are preserved (the pre-existing `replace_active_history`
// clobbered the entire session, which broke multi-topic isolation).
tx.execute(
"DELETE FROM messages WHERE session_id = ?1 AND topic_id = ?2",
params![session_id, topic_id],
)?;
// Continue seq from the session-wide max so we don't violate
// UNIQUE(session_id, seq). Other topics' messages keep their seqs.
let start_seq: i64 = tx.query_row(
"SELECT COALESCE(MAX(seq), 0) + 1 FROM messages WHERE session_id = ?1",
params![session_id],
|row| row.get(0),
)?;
for (i, message) in messages.iter().enumerate() {
let seq = start_seq + i as i64;
insert_message_with_topic_seq(&tx, session_id, topic_id, seq, message)?;
}
// Update this topic's message_count and timestamps.
tx.execute(
"UPDATE topics SET message_count = ?2, last_active_at = ?3, updated_at = ?3 WHERE id = ?1",
params![topic_id, messages.len() as i64, now],
)?;
// Recompute session-wide counts from the messages table so they stay
// consistent after a partial replacement (we only touched one topic,
// so we can't just set the session count to `messages.len()`).
let (total_count, user_turn_count): (i64, i64) = tx.query_row(
"SELECT COUNT(*), COALESCE(SUM(CASE WHEN role = 'user' THEN 1 ELSE 0 END), 0) \
FROM messages WHERE session_id = ?1",
params![session_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
tx.execute(
"UPDATE sessions SET message_count = ?2, user_turn_count = ?3, \
updated_at = ?4, last_active_at = ?4, archived_at = NULL \
WHERE id = ?1 AND deleted_at IS NULL",
params![session_id, total_count, user_turn_count, now],
)?;
tx.commit()?;
Ok(())
}
pub fn mark_agent_prompt_reinjected(&self, session_id: &str) -> Result<(), StorageError> {
let now = current_timestamp();
let conn = self.pool.get()?;
@ -1667,6 +1735,51 @@ fn insert_message_with_seq(
Ok(())
}
/// Insert a message with an explicit `topic_id` and `seq`.
///
/// Used by `replace_topic_history` to insert compressed messages while
/// preserving topic association (the plain `insert_message_with_seq` would
/// set topic_id to NULL).
fn insert_message_with_topic_seq(
conn: &rusqlite::Transaction<'_>,
session_id: &str,
topic_id: &str,
seq: i64,
message: &ChatMessage,
) -> Result<(), StorageError> {
let media_refs_json = serde_json::to_string(&message.media_refs)?;
let tool_calls_json = message
.tool_calls
.as_ref()
.map(serde_json::to_string)
.transpose()?;
conn.execute(
"
INSERT INTO messages (
id, session_id, topic_id, seq, role, content,
system_context, reasoning_content, media_refs_json, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)
",
params![
message.id,
session_id,
topic_id,
seq,
message.role,
message.content,
message.system_context,
message.reasoning_content,
media_refs_json,
message.tool_call_id,
message.tool_name,
tool_calls_json,
message.tool_duration_ms.map(|v| v as i64),
message.timestamp,
],
)?;
Ok(())
}
fn clone_message_for_compaction(message: &ChatMessage, timestamp: i64) -> ChatMessage {
ChatMessage {
id: uuid::Uuid::new_v4().to_string(),

View File

@ -63,6 +63,17 @@ pub trait ConversationRepository: Send + Sync + 'static {
session_id: &str,
messages: &[ChatMessage],
) -> Result<(), StorageError>;
/// Replace the entire history for a specific topic.
/// Deletes only messages belonging to the given topic_id, then inserts
/// the new messages with topic_id set correctly. Used by compressor when
/// it has produced a complete, validated message list for a single topic.
fn replace_topic_history(
&self,
session_id: &str,
topic_id: &str,
messages: &[ChatMessage],
) -> Result<(), StorageError>;
}
pub trait PromptInjectionRepository: Send + Sync + 'static {
@ -252,6 +263,15 @@ impl ConversationRepository for super::SessionStore {
) -> Result<(), StorageError> {
super::SessionStore::replace_active_history(self, session_id, messages)
}
fn replace_topic_history(
&self,
session_id: &str,
topic_id: &str,
messages: &[ChatMessage],
) -> Result<(), StorageError> {
super::SessionStore::replace_topic_history(self, session_id, topic_id, messages)
}
}
impl PromptInjectionRepository for super::SessionStore {

View File

@ -33,11 +33,31 @@ impl FileEditTool {
// Check directory restriction
if let Some(ref allowed) = self.allowed_dir {
let allowed_path = Path::new(allowed);
if !resolved.starts_with(allowed_path) {
// canonicalize both paths to resolve symlinks and prevent path traversal
// via symlinks inside allowed_dir pointing outside.
// For edit tool the target file may not exist yet; fall back to
// canonicalizing the parent directory.
let canonical_allowed = std::fs::canonicalize(allowed)
.map_err(|e| format!("Failed to canonicalize allowed dir '{}': {}", allowed, e))?;
let canonical_resolved = match std::fs::canonicalize(&resolved) {
Ok(c) => c,
Err(_) => {
// File doesn't exist yet; canonicalize parent directory
let parent = resolved.parent().ok_or_else(|| {
format!("Path '{}' has no parent directory", path)
})?;
let canonical_parent = std::fs::canonicalize(parent).map_err(|e| {
format!("Failed to canonicalize parent directory of '{}': {}", path, e)
})?;
canonical_parent.join(resolved.file_name().ok_or_else(|| {
format!("Path '{}' has no file name component", path)
})?)
}
};
if !canonical_resolved.starts_with(&canonical_allowed) {
return Err(format!(
"Path '{}' is outside allowed directory '{}'",
path, allowed
"Path '{}' (resolves to '{}') is outside allowed directory '{}'",
path, canonical_resolved.display(), canonical_allowed.display()
));
}
}

View File

@ -37,11 +37,18 @@ impl FileReadTool {
// Check directory restriction
if let Some(ref allowed) = self.allowed_dir {
let allowed_path = Path::new(allowed);
if !resolved.starts_with(allowed_path) {
// canonicalize both paths to resolve symlinks and prevent path traversal
// via symlinks inside allowed_dir pointing outside
let canonical_allowed = std::fs::canonicalize(allowed)
.map_err(|e| format!("Failed to canonicalize allowed dir '{}': {}", allowed, e))?;
// For read tool, file must exist; canonicalize will fail for non-existent paths
// which is acceptable (returns error)
let canonical_resolved = std::fs::canonicalize(&resolved)
.map_err(|e| format!("Failed to canonicalize path '{}': {}", path, e))?;
if !canonical_resolved.starts_with(&canonical_allowed) {
return Err(format!(
"Path '{}' is outside allowed directory '{}'",
path, allowed
"Path '{}' (resolves to '{}') is outside allowed directory '{}'",
path, canonical_resolved.display(), canonical_allowed.display()
));
}
}

View File

@ -32,11 +32,31 @@ impl FileWriteTool {
// Check directory restriction
if let Some(ref allowed) = self.allowed_dir {
let allowed_path = Path::new(allowed);
if !resolved.starts_with(allowed_path) {
// canonicalize both paths to resolve symlinks and prevent path traversal
// via symlinks inside allowed_dir pointing outside.
// For write tool the target file may not exist yet; fall back to
// canonicalizing the parent directory.
let canonical_allowed = std::fs::canonicalize(allowed)
.map_err(|e| format!("Failed to canonicalize allowed dir '{}': {}", allowed, e))?;
let canonical_resolved = match std::fs::canonicalize(&resolved) {
Ok(c) => c,
Err(_) => {
// File doesn't exist yet; canonicalize parent directory
let parent = resolved.parent().ok_or_else(|| {
format!("Path '{}' has no parent directory", path)
})?;
let canonical_parent = std::fs::canonicalize(parent).map_err(|e| {
format!("Failed to canonicalize parent directory of '{}': {}", path, e)
})?;
canonical_parent.join(resolved.file_name().ok_or_else(|| {
format!("Path '{}' has no file name component", path)
})?)
}
};
if !canonical_resolved.starts_with(&canonical_allowed) {
return Err(format!(
"Path '{}' is outside allowed directory '{}'",
path, allowed
"Path '{}' (resolves to '{}') is outside allowed directory '{}'",
path, canonical_resolved.display(), canonical_allowed.display()
));
}
}

View File

@ -311,6 +311,7 @@ impl Tool for HttpRequestTool {
let client = match reqwest::Client::builder()
.timeout(Duration::from_secs(self.timeout_secs))
.redirect(reqwest::redirect::Policy::none())
.build()
{
Ok(c) => c,

View File

@ -87,6 +87,21 @@ impl ToolRegistry {
*new_registry.tools.write().expect("ToolRegistry lock poisoned") = filtered;
new_registry
}
/// 创建一个仅包含指定工具的新 registry 副本(白名单)。
/// include 中不存在于当前 registry 的名称会被静默跳过(取交集语义)。
pub fn only(&self, include: &[&str]) -> Self {
let include_set: std::collections::HashSet<&str> = include.iter().copied().collect();
let tools = self.tools.read().expect("ToolRegistry lock poisoned");
let filtered: HashMap<String, Arc<dyn ToolTrait>> = tools
.iter()
.filter(|(name, _)| include_set.contains(name.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let new_registry = ToolRegistry::new();
*new_registry.tools.write().expect("ToolRegistry lock poisoned") = filtered;
new_registry
}
}
impl Default for ToolRegistry {
@ -94,3 +109,82 @@ impl Default for ToolRegistry {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::tools::traits::ToolResult;
use async_trait::async_trait;
/// 仅用于测试的占位工具,按构造名注册
struct FakeTool {
tool_name: String,
}
#[async_trait]
impl ToolTrait for FakeTool {
fn name(&self) -> &str {
&self.tool_name
}
fn description(&self) -> &str {
"fake"
}
fn parameters_schema(&self) -> serde_json::Value {
serde_json::json!({})
}
async fn execute(&self, _args: serde_json::Value) -> anyhow::Result<ToolResult> {
Ok(ToolResult {
success: true,
output: String::new(),
error: None,
})
}
}
fn registry_with(names: &[&str]) -> ToolRegistry {
let reg = ToolRegistry::new();
for n in names {
reg.register(FakeTool {
tool_name: n.to_string(),
});
}
reg
}
fn sorted_names(reg: &ToolRegistry) -> Vec<String> {
let mut v = reg.tool_names();
v.sort();
v
}
#[test]
fn only_keeps_listed_tools() {
let reg = registry_with(&["read", "edit", "write", "bash"]);
let filtered = reg.only(&["read", "bash"]);
assert_eq!(sorted_names(&filtered), vec!["bash", "read"]);
}
#[test]
fn only_silently_skips_missing_names() {
let reg = registry_with(&["read", "edit"]);
let filtered = reg.only(&["read", "nonexistent", "glob"]);
assert_eq!(sorted_names(&filtered), vec!["read"]);
}
#[test]
fn only_with_empty_include_returns_empty() {
let reg = registry_with(&["read", "edit"]);
let filtered = reg.only(&[]);
assert!(filtered.tool_names().is_empty());
}
#[test]
fn only_does_not_mutate_source() {
let reg = registry_with(&["read", "edit", "write"]);
let _ = reg.only(&["read"]);
// 源 registry 不受影响
let mut v = reg.tool_names();
v.sort();
assert_eq!(v, vec!["edit", "read", "write"]);
}
}

View File

@ -102,6 +102,7 @@ mod tests {
prompt_template: "任务: {{description}}\n指令: {{prompt}}".to_string(),
body: None,
allowed_tools: None,
denied_tools: None,
max_execution_secs: None,
source: SubagentSource::Builtin,
path: None,

View File

@ -29,8 +29,6 @@ pub struct SubAgentRuntimeConfig {
pub default_allowed_tools: HashSet<String>,
/// 默认最大执行时间(秒)
pub default_max_execution_secs: u64,
/// Explore 类型的最大执行时间(秒)
pub explore_max_execution_secs: u64,
/// 任务 TTL小时
pub ttl_hours: u64,
/// 技能索引(可选,预生成的技能列表字符串)
@ -57,7 +55,6 @@ impl Default for SubAgentRuntimeConfig {
"send_session_message".to_string(), // 用于进度通知
]),
default_max_execution_secs: 3600, // 60分钟
explore_max_execution_secs: 3600, // 60分钟
ttl_hours: 24,
skills_index: None,
max_nesting_depth: 1,
@ -394,23 +391,80 @@ impl DefaultSubAgentRuntime {
.unwrap_or(self.config.default_max_execution_secs)
}
/// 根据 def 与嵌套深度构建子代理工具集。
/// 过滤顺序base → allowed_tools 白名单 → denied_tools 黑名单 + depth 达到上限移除 task。
/// - `allowed_tools` 为 Some 时取交集白名单None 表示不限制。
/// - `denied_tools` 为 Some 时扣除(黑名单),在白名单之后应用。
/// - 当 child_depth >= max_nesting_depth 时移除 task 工具(防无限嵌套的安全兜底,
/// 不可被 def 覆盖)。默认 max_nesting_depth=2即孙代理depth=2无法再创建子代理。
fn build_subagent_tools_registry(
&self,
def: Option<&SubagentDef>,
child_depth: u32,
) -> Arc<ToolRegistry> {
let depth_deny_task = child_depth >= self.config.max_nesting_depth;
let allowed: Option<&Vec<String>> = def.and_then(|d| d.allowed_tools.as_ref());
let denied_tools: Option<&Vec<String>> = def.and_then(|d| d.denied_tools.as_ref());
// 快速路径:无白名单、无黑名单、无需 depth 兜底 → 直接复用 Arc避免拷贝
if allowed.is_none() && denied_tools.is_none() && !depth_deny_task {
return self.subagent_tools.clone();
}
Arc::new(Self::filter_tool_registry(
&self.subagent_tools,
allowed,
denied_tools,
depth_deny_task,
))
}
/// 纯函数:在 base 之上应用白名单/黑名单/depth 规则。
/// 抽取为关联函数便于单元测试(无需构造整个 DefaultSubAgentRuntime
fn filter_tool_registry(
base: &ToolRegistry,
allowed: Option<&Vec<String>>,
denied_tools: Option<&Vec<String>>,
depth_deny_task: bool,
) -> ToolRegistry {
// 1. 应用白名单(若存在),否则取得 owned 副本以便后续黑名单过滤
let tools: ToolRegistry = match allowed {
Some(list) => {
let refs: Vec<&str> = list.iter().map(|s| s.as_str()).collect();
base.only(&refs)
}
None => base.without(&[]),
};
// 2. 合并黑名单depth 规则 + denied_tools
let mut denied: Vec<&str> = Vec::new();
if depth_deny_task {
denied.push(TaskTool::TOOL_NAME);
}
if let Some(dt) = denied_tools {
denied.extend(dt.iter().map(|s| s.as_str()));
}
if denied.is_empty() {
tools
} else {
tools.without(&denied)
}
}
/// 创建子代理实例
fn create_subagent(
&self,
session: &TaskSession,
system_prompt: String,
def: Option<&SubagentDef>,
parent_nesting_depth: u32,
parent_task_id: Option<String>,
) -> Result<AgentLoop, TaskError> {
let prompt_provider = Arc::new(StaticSystemPromptProvider::new(system_prompt));
// 孙智能体depth >= 2不注册 task 工具,防止无限嵌套
let child_depth = parent_nesting_depth + 1;
let tools = if child_depth >= 2 {
Arc::new(self.subagent_tools.without(&[TaskTool::TOOL_NAME]))
} else {
self.subagent_tools.clone()
};
let tools = self.build_subagent_tools_registry(def, child_depth);
AgentLoop::with_tools_and_system_prompt_provider(
AgentRuntimeConfig::from(self.provider_config.clone()),
@ -481,11 +535,7 @@ impl DefaultSubAgentRuntime {
};
// 设置超时
let max_secs = if session.subagent_type == "explore" {
self.config.explore_max_execution_secs
} else {
self.effective_max_execution_secs(def)
};
let max_secs = self.effective_max_execution_secs(def);
let timeout_duration = Duration::from_secs(max_secs);
let result = tokio::time::timeout(
@ -663,7 +713,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
);
// 7. 创建子代理
let agent = self.create_subagent(&session, system_prompt, parent_context.nesting_depth, parent_context.task_id.clone())?;
let agent = self.create_subagent(&session, system_prompt, Some(&def), parent_context.nesting_depth, parent_context.task_id.clone())?;
// 8. 执行任务
let result = self
@ -747,8 +797,16 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
&additional_prompt,
);
// 4.1 重新解析 def 以应用工具过滤。
// 安全要求def 被删除/禁用时必须失败恢复,而不是降级为完整工具集——
// 否则一个受限子代理(如 allowed_tools: [read])在 def 失踪后会获得全部工具,
// 构成权限提升。与 spawn 保持一致def 不可用即拒绝执行。
let def = self
.find_subagent_def(&session.subagent_type)
.map_err(TaskError::InvalidArguments)?;
// 5. 创建子代理
let agent = self.create_subagent(&session, system_prompt, parent_context.nesting_depth, parent_context.task_id.clone())?;
let agent = self.create_subagent(&session, system_prompt, Some(&def), parent_context.nesting_depth, parent_context.task_id.clone())?;
// 6. 使用历史继续执行
let result = self
@ -809,7 +867,6 @@ impl SubagentCatalog {
pub fn new() -> Self {
let mut catalog = Self::default();
catalog.register(SubagentDef::builtin_general());
catalog.register(SubagentDef::builtin_explore());
catalog
}
@ -826,7 +883,6 @@ impl SubagentCatalog {
// 先内置作为基础
let mut merged: std::collections::HashMap<String, SubagentDef> = std::collections::HashMap::new();
merged.insert("general".to_string(), SubagentDef::builtin_general());
merged.insert("explore".to_string(), SubagentDef::builtin_explore());
tracing::debug!(cwd = %cwd.display(), "Discovering subagents from cwd");
@ -960,6 +1016,12 @@ pub struct SubagentWithStatus {
pub source: String,
/// Which scopes have this subagent disabled. Empty means enabled.
pub disabled_in_scopes: Vec<String>,
/// 工具白名单None 表示不过滤)
#[serde(default, skip_serializing_if = "Option::is_none")]
pub allowed_tools: Option<Vec<String>>,
/// 工具黑名单None 表示不过滤)
#[serde(default, skip_serializing_if = "Option::is_none")]
pub denied_tools: Option<Vec<String>>,
}
#[derive(Debug, Clone)]
@ -1105,6 +1167,8 @@ impl SubagentRuntime {
description: def.description.clone(),
source: def.source.as_str().to_string(),
disabled_in_scopes: scopes.iter().map(|s| s.as_str().to_string()).collect(),
allowed_tools: def.allowed_tools.clone(),
denied_tools: def.denied_tools.clone(),
}
})
.collect();
@ -1341,6 +1405,8 @@ struct SubagentFrontmatter {
#[serde(default)]
allowed_tools: Option<Vec<String>>,
#[serde(default)]
denied_tools: Option<Vec<String>>,
#[serde(default)]
max_execution_secs: Option<u64>,
}
@ -1401,11 +1467,14 @@ fn parse_subagent_file(path: &Path, source: SubagentSource) -> Result<SubagentDe
let content = fs::read_to_string(path)
.map_err(|e| format!("failed to read file: {}", e))?;
let (frontmatter_raw, body) = split_frontmatter(&content)
.ok_or_else(|| "missing YAML frontmatter block".to_string())?;
let frontmatter: SubagentFrontmatter = serde_yaml::from_str(frontmatter_raw)
.map_err(|e| format!("invalid YAML frontmatter: {}", e))?;
let (frontmatter, body) = match crate::frontmatter::parse::<SubagentFrontmatter>(&content) {
Ok(v) => v,
Err(err) => {
let bytes = content.len();
let crlf = content.contains('\r');
return Err(format!("{} (bytes={}, crlf={})", err, bytes, crlf));
}
};
if frontmatter.description.trim().is_empty() {
return Err("description is required and cannot be empty".to_string());
@ -1428,43 +1497,13 @@ fn parse_subagent_file(path: &Path, source: SubagentSource) -> Result<SubagentDe
prompt_template,
body: if body_content.is_empty() { None } else { Some(body_content) },
allowed_tools: frontmatter.allowed_tools,
denied_tools: frontmatter.denied_tools,
max_execution_secs: frontmatter.max_execution_secs,
source,
path: Some(path.to_path_buf()),
})
}
/// 分割 frontmatter 和 body
fn split_frontmatter(content: &str) -> Option<(&str, &str)> {
// 跳过开头的 ---
let content = content
.strip_prefix("---")
.or_else(|| content.strip_prefix("---"))?;
// 跳过 --- 后的换行符和可能的空行
let content = content.trim_start_matches('\r').trim_start_matches('\n');
// 找结束标记(容忍不同的换行符格式和前面的空行)
// 尝试多种可能的结束标记格式
let end_markers = ["\n---\n", "\n---", "\r\n---\r\n", "\r\n---"];
let mut idx = None;
let mut marker_len = 0;
for marker in end_markers {
if let Some(pos) = content.find(marker) {
idx = Some(pos);
marker_len = marker.len();
break;
}
}
let idx = idx?;
let frontmatter = &content[..idx];
let body = &content[idx + marker_len..];
let body = body.trim_start_matches('\r').trim_start_matches('\n');
Some((frontmatter, body))
}
#[cfg(test)]
mod tests {
use super::*;
@ -1538,10 +1577,9 @@ mod tests {
assert!(change.changed);
assert!(!change.available);
// 禁用后 prompt 不应包含 generalexplore 仍可用,所以 prompt 仍为 Some
let prompt = runtime.system_index_prompt_filtered().unwrap();
assert!(!prompt.contains("<name>general</name>"));
assert!(prompt.contains("<name>explore</name>"));
// 禁用后 prompt 不应包含 general无可用子代理时返回 None
let prompt = runtime.system_index_prompt_filtered();
assert!(prompt.map_or(true, |p| !p.contains("<name>general</name>")));
}
#[test]
@ -1556,8 +1594,8 @@ mod tests {
runtime
.disable_subagent(SubagentScope::Project, "general")
.unwrap();
let prompt = runtime.system_index_prompt_filtered().unwrap();
assert!(!prompt.contains("<name>general</name>"));
// 无可用子代理时返回 None
assert!(runtime.system_index_prompt_filtered().is_none());
let change = runtime
.enable_subagent(SubagentScope::Project, "general")
@ -1586,10 +1624,6 @@ mod tests {
assert!(general
.disabled_in_scopes
.contains(&"project".to_string()));
// explore 应仍启用
let explore = items.iter().find(|i| i.name == "explore").unwrap();
assert!(explore.disabled_in_scopes.is_empty());
}
#[test]
@ -1605,12 +1639,10 @@ mod tests {
.unwrap();
assert!(runtime.find_available("general").is_none());
assert!(runtime.find_available("explore").is_some());
// available_names 不应包含 general
let names = runtime.available_names();
assert!(!names.contains(&"general".to_string()));
assert!(names.contains(&"explore".to_string()));
}
#[test]
@ -1626,4 +1658,225 @@ mod tests {
.unwrap_err();
assert!(err.contains("not found"));
}
// ===== 工具过滤allowed_tools / denied_tools测试 =====
use crate::tools::traits::{Tool as ToolTrait, ToolResult};
/// 占位工具,按构造名注册
struct FakeTool {
tool_name: String,
}
#[async_trait::async_trait]
impl ToolTrait for FakeTool {
fn name(&self) -> &str {
&self.tool_name
}
fn description(&self) -> &str {
"fake"
}
fn parameters_schema(&self) -> serde_json::Value {
serde_json::json!({})
}
async fn execute(&self, _args: serde_json::Value) -> anyhow::Result<ToolResult> {
Ok(ToolResult {
success: true,
output: String::new(),
error: None,
})
}
}
fn base_registry() -> ToolRegistry {
let reg = ToolRegistry::new();
for name in &["read", "edit", "write", "bash", "task"] {
reg.register(FakeTool {
tool_name: name.to_string(),
});
}
reg
}
fn sorted_names(reg: &ToolRegistry) -> Vec<String> {
let mut v = reg.tool_names();
v.sort();
v
}
/// 把 &str 切片转为 Some(Vec<String>),便于构造过滤参数
fn s(v: &[&str]) -> Option<Vec<String>> {
Some(v.iter().map(|x| x.to_string()).collect())
}
#[test]
fn filter_no_restriction_returns_all() {
let base = base_registry();
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, None, None, false);
assert_eq!(
sorted_names(&reg),
vec!["bash", "edit", "read", "task", "write"]
);
}
#[test]
fn filter_depth_deny_task_removes_task() {
let base = base_registry();
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, None, None, true);
assert_eq!(
sorted_names(&reg),
vec!["bash", "edit", "read", "write"]
);
}
#[test]
fn filter_whitelist_keeps_only_listed() {
let base = base_registry();
let allowed = s(&["read", "bash"]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, allowed.as_ref(), None, false);
assert_eq!(sorted_names(&reg), vec!["bash", "read"]);
}
#[test]
fn filter_whitelist_skips_missing_names() {
let base = base_registry();
// 包含未注册的工具名应被静默跳过
let allowed = s(&["read", "nonexistent", "glob"]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, allowed.as_ref(), None, false);
assert_eq!(sorted_names(&reg), vec!["read"]);
}
#[test]
fn filter_blacklist_removes_listed() {
let base = base_registry();
let denied = s(&["bash", "task"]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, None, denied.as_ref(), false);
assert_eq!(sorted_names(&reg), vec!["edit", "read", "write"]);
}
#[test]
fn filter_whitelist_then_blacklist() {
let base = base_registry();
let allowed = s(&["read", "bash"]);
let denied = s(&["bash"]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(
&base,
allowed.as_ref(),
denied.as_ref(),
false,
);
// 白名单留下 read+bash黑名单再扣除 bash
assert_eq!(sorted_names(&reg), vec!["read"]);
}
#[test]
fn filter_empty_whitelist_yields_empty() {
let base = base_registry();
let allowed = s(&[]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, allowed.as_ref(), None, false);
assert!(reg.tool_names().is_empty());
}
#[test]
fn filter_depth_rule_overrides_whitelist_task() {
let base = base_registry();
// 白名单显式包含 task但 depth≥2 安全兜底仍应移除它
let allowed = s(&["read", "task"]);
let reg = DefaultSubAgentRuntime::filter_tool_registry(&base, allowed.as_ref(), None, true);
assert_eq!(sorted_names(&reg), vec!["read"]);
}
// ===== frontmatter 解析denied_tools测试 =====
#[test]
fn parse_subagent_file_handles_crlf_endings() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("SUBAGENT.md");
std::fs::write(
&path,
"---\r\nname: demo\r\ndescription: demo subagent\r\n---\r\nStep A\r\nStep B",
)
.unwrap();
let subagent = parse_subagent_file(&path, SubagentSource::Project).unwrap();
assert_eq!(subagent.name, "demo");
assert_eq!(subagent.description, "demo subagent");
assert_eq!(subagent.body.as_deref(), Some("Step A\nStep B"));
}
#[test]
fn parse_subagent_file_reads_denied_tools() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("SUBAGENT.md");
std::fs::write(
&path,
"---\n\
name: sandbox\n\
description: sandbox agent\n\
allowed_tools: [read, todo_write]\n\
denied_tools: [bash, task]\n\
---\n\
body instructions",
)
.unwrap();
let def = parse_subagent_file(&path, SubagentSource::Project).unwrap();
assert_eq!(def.name, "sandbox");
assert_eq!(
def.allowed_tools.as_deref(),
Some(["read".to_string(), "todo_write".to_string()].as_slice())
);
assert_eq!(
def.denied_tools.as_deref(),
Some(["bash".to_string(), "task".to_string()].as_slice())
);
}
#[test]
fn parse_subagent_file_denied_tools_default_none_when_absent() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("SUBAGENT.md");
std::fs::write(
&path,
"---\nname: basic\ndescription: basic agent\n---\nbody",
)
.unwrap();
let def = parse_subagent_file(&path, SubagentSource::User).unwrap();
assert!(def.allowed_tools.is_none());
assert!(def.denied_tools.is_none());
}
#[test]
fn list_with_status_projects_tool_fields() {
let temp = tempfile::tempdir().unwrap();
let mut catalog = SubagentCatalog::new();
catalog.register(SubagentDef {
name: "sandbox".to_string(),
description: "sandbox agent".to_string(),
prompt_template: String::new(),
body: None,
allowed_tools: Some(vec!["read".to_string(), "todo_write".to_string()]),
denied_tools: Some(vec!["bash".to_string()]),
max_execution_secs: None,
source: SubagentSource::Builtin,
path: None,
});
let runtime = SubagentRuntime::new(
SubagentsConfig::default(),
Arc::new(catalog),
temp.path().to_path_buf(),
);
let items = runtime.list_with_status();
let item = items.iter().find(|i| i.name == "sandbox").unwrap();
assert_eq!(
item.allowed_tools.as_deref(),
Some(["read".to_string(), "todo_write".to_string()].as_slice())
);
assert_eq!(
item.denied_tools.as_deref(),
Some(["bash".to_string()].as_slice())
);
}
}

View File

@ -60,8 +60,10 @@ pub struct SubagentDef {
pub prompt_template: String,
/// 可选的详细指令body 部分)
pub body: Option<String>,
/// 工具白名单None 表示使用默认
/// 工具白名单None 表示不过滤Some 时仅这些工具可用
pub allowed_tools: Option<Vec<String>>,
/// 工具黑名单None 表示不过滤Some 时这些工具被禁用;在白名单之后应用)
pub denied_tools: Option<Vec<String>>,
/// 最大执行时间None 表示使用默认
pub max_execution_secs: Option<u64>,
/// 来源
@ -79,20 +81,7 @@ impl SubagentDef {
prompt_template: "你是一个专注的子代理,正在执行一个独立任务。\n\n任务描述: {{description}}\n\n你应该:\n1. 专注于完成任务,不要偏离目标\n2. 使用可用的工具进行必要操作\n3. 完成后给出简洁的总结\n4. 不要尝试创建新的子代理任务\n\n任务追踪:\n你可以使用 `todo_write` 工具追踪子任务进度。规则:同一时间只有一个 in_progress完成后再标记下一个3步以上才使用。\n\n注意: 你没有访问主对话历史的权限,这是一个独立的执行上下文。".to_string(),
body: None,
allowed_tools: None,
max_execution_secs: None,
source: SubagentSource::Builtin,
path: None,
}
}
/// 创建内置 explore 子代理定义
pub fn builtin_explore() -> Self {
Self {
name: "explore".to_string(),
description: "探索型子代理 - 只读搜索代理".to_string(),
prompt_template: "你是一个只读探索代理,用于代码库探索和信息收集。\n\n任务描述: {{description}}\n\n你应该:\n1. 只使用只读工具进行探索\n2. 专注于理解和收集信息\n3. 不要进行任何写操作\n4. 给出简洁的发现总结\n\n注意: 你是一个只读代理,禁止执行任何修改操作。".to_string(),
body: None,
allowed_tools: None,
denied_tools: None,
max_execution_secs: None,
source: SubagentSource::Builtin,
path: None,

View File

@ -526,7 +526,6 @@ export function ConfigPage({ onClose, onSaveConnection, initialTab }: ConfigPage
<SectionCard title="Task 子代理">
<div className="flex items-center justify-between"><span className="text-sm text-[var(--text-secondary)]"> Task </span><Toggle checked={config.tools.task.enabled} onChange={v => update('tools', { ...config.tools, task: { ...config.tools.task, enabled: v } })} /></div>
<Field label="最大执行时间 (秒)"><input type="number" value={config.tools.task.max_execution_secs} onChange={e => update('tools', { ...config.tools, task: { ...config.tools.task, max_execution_secs: +e.target.value } })} className={inputCls} /></Field>
<Field label="探索模式最大执行时间 (秒)"><input type="number" value={config.tools.task.explore_max_execution_secs} onChange={e => update('tools', { ...config.tools, task: { ...config.tools.task, explore_max_execution_secs: +e.target.value } })} className={inputCls} /></Field>
<Field label="TTL (小时)"><input type="number" value={config.tools.task.ttl_hours} onChange={e => update('tools', { ...config.tools, task: { ...config.tools.task, ttl_hours: +e.target.value } })} className={inputCls} /></Field>
</SectionCard>
<SectionCard title="允许的工具列表">
@ -612,6 +611,31 @@ export function ConfigPage({ onClose, onSaveConnection, initialTab }: ConfigPage
}
}
const toolLabel = (key: string): string => {
const known = TASK_KNOWN_TOOLS.find(t => t.key === key)
return known ? known.label : key
}
const renderToolTags = (
label: string,
tools: string[] | undefined,
tone: 'allow' | 'deny',
) => {
if (!tools || tools.length === 0) return null
const tagCls =
tone === 'allow'
? 'bg-emerald-500/10 text-emerald-600 dark:text-emerald-400'
: 'bg-rose-500/10 text-rose-600 dark:text-rose-400'
return (
<div className="flex items-center gap-1 flex-wrap mt-1">
<span className="text-[10px] text-[var(--text-muted)]">{label}:</span>
{tools.map(t => (
<span key={t} className={`text-[10px] px-1.5 py-0.5 rounded ${tagCls}`}>{toolLabel(t)}</span>
))}
</div>
)
}
return (
<SectionCard title="已发现子代理" subtitle="即时生效">
{subagentListLoading && subagents.length === 0 ? (
@ -632,6 +656,8 @@ export function ConfigPage({ onClose, onSaveConnection, initialTab }: ConfigPage
<span className="text-[10px] px-1.5 py-0.5 rounded bg-[var(--bg-tertiary)] text-[var(--text-muted)] uppercase tracking-wider">{subagent.source}</span>
</div>
<p className="text-xs text-[var(--text-muted)] truncate mt-0.5">{subagent.description}</p>
{renderToolTags('允许', subagent.allowed_tools, 'allow')}
{renderToolTags('禁用', subagent.denied_tools, 'deny')}
</div>
<Toggle checked={isEnabled} onChange={() => handleToggle(subagent.name, isEnabled)} />
</div>

View File

@ -7,7 +7,7 @@ export interface GatewayConfig { host: string; port: number; show_tool_results:
export interface TimeConfig { timezone: string }
export interface SchedulerConfig { enabled: boolean; tick_resolution_ms: number; worker_queue_capacity: number; misfire_policy: 'skip' | 'catch_up'; jobs?: SchedulerJobConfig[] }
export interface SkillsConfig { enabled: boolean; sources: string[]; max_index_chars: number; max_listed_skills: number }
export interface TaskConfig { enabled: boolean; max_execution_secs: number; explore_max_execution_secs: number; ttl_hours: number; allowed_tools: string[] }
export interface TaskConfig { enabled: boolean; max_execution_secs: number; ttl_hours: number; allowed_tools: string[] }
export interface ToolsConfig { disabled: string[]; task: TaskConfig }
export interface MemoryMaintenanceConfig { max_merge_ratio: number; min_memories_to_keep: number; max_merge_per_group: number }
export interface ImageContextConfig { max_images_in_context: number; max_image_age_rounds: number }
@ -45,6 +45,8 @@ export interface SubagentItem {
description: string
source: string
disabled_in_scopes: string[]
allowed_tools?: string[]
denied_tools?: string[]
}
export interface SubagentListResponse {