ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

feat: 添加 todo_write 工具，支持全量替换和增量合并两种模式

Browse Source 
- Tool: 纯内存实现 (Arc<RwLock<HashMap>>)，零 DB 依赖，解耦持久化
- 状态机: pending → in_progress → completed/cancelled，单 in_progress 约束
- merge=false: 全量替换模式（默认）
- merge=true: 增量更新模式，只传变更的项，其余保留
- 隔离: scope_key = topic_id.unwrap_or(session_id)，topic 和子代理隔离
- 持久化: TodoRepository trait + SessionStore SQLite 实现，在 Session 拦截器层完成
- 前端推送: WsOutbound::TodoList 事件
- Prompt: TodoPromptProvider 中文指令，子代理模板也包含
- 测试: 16 个单元测试，全部通过

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
oudecheng 2026-06-12 14:19:07 +08:00
parent cedd8b2a69
commit 881fcace47
15 changed files with 1508 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/gateway/agent_factory.rs
View File

@ -3,6 +3,7 @@ use std::sync::Arc;
use crate::agent::{AgentError, AgentLoop, CompositeSystemPromptProvider};
use crate::config::LLMProviderConfig;
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
use crate::gateway::todo_prompt_provider::TodoPromptProvider;
use crate::skills::{SkillPromptProvider, SkillRuntime};
use crate::storage::persistent_session_id;
use crate::storage::PromptInjectionRepository;
@ -53,6 +54,7 @@ impl AgentFactory {
self.prompt_repository.clone(),
)),
Box::new(SkillPromptProvider::new(self.skills.clone())),
Box::new(TodoPromptProvider::new()),
]));
AgentLoop::with_tools_and_system_prompt_provider(

8
src/gateway/execution.rs
View File

@ -190,6 +190,14 @@ impl AgentExecutionService {
// 只有当是最新回合时才触发历史压缩
let should_schedule_compaction = is_current_turn;
// 拦截 todo_write 结果：持久化 + 前端推送
if is_current_turn {
session.intercept_todo_write_results(
&request.result.emitted_messages,
request.chat_id,
);
}
Ok(FinalizedAgentResult {
outbound_messages,
should_schedule_compaction,

1
src/gateway/mod.rs
View File

@ -25,6 +25,7 @@ pub mod session_message_service;
pub mod session_pool;
pub mod static_files;
pub mod tool_registry_factory;
pub mod todo_prompt_provider;
pub mod ws;
use axum::{Router, routing};

8
src/gateway/runtime.rs
View File

@ -3,6 +3,8 @@
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::agent::AgentError;
use crate::bus::MessageBus;
use crate::config::{LLMProviderConfig, MemoryMaintenanceConfig, SubagentsConfig, TaskConfig};
@ -18,6 +20,7 @@ use crate::tools::{
SessionMessageSender, SubAgentRuntimeConfig, SubagentCatalog, ToolRegistry,
};
use crate::tools::task::repository::TaskRepository;
use crate::tools::todo_write::TodoItem;
use super::agent_factory::AgentFactory;
use super::cli_session::CliSessionService;
@ -117,6 +120,11 @@ pub(crate) fn build_session_manager_with_sender(
task_config.clone(),
);
// Create shared todo state for TodoWriteTool
let todo_state: Arc<RwLock<HashMap<String, Vec<TodoItem>>>> =
Arc::new(RwLock::new(HashMap::new()));
let factory = factory.with_todo_state(todo_state);
// Create MCP Initializer (async, non-blocking)
// MCP servers connect in background task
let mut mcp_initializer = McpInitializer::with_config(mcp_config);

78
src/gateway/session.rs
View File

@ -385,6 +385,84 @@ impl Session {
let _ = self.user_tx.send(msg).await;
}
/// 扫描 agent 结果中的 todo_write 工具消息，
/// 提取 todos 并做持久化 + 前端推送（同步版本）。
pub(crate) fn intercept_todo_write_results(
&self,
emitted_messages: &[ChatMessage],
chat_id: &str,
) {
for msg in emitted_messages {
if msg.role != "tool" {
continue;
}
if msg.tool_name.as_deref() != Some("todo_write") {
continue;
}
// 解析工具返回的 JSON
let parsed: serde_json::Value = match serde_json::from_str(&msg.content) {
Ok(v) => v,
Err(_) => continue,
};
let Some(todos_array) = parsed
.get("current_todos")
.and_then(|v| v.as_array())
else {
continue;
};
// 计算持久化所需的 key
let session_id = crate::storage::persistent_session_id(&self.channel_name, chat_id);
let topic_id = self.current_topic(chat_id);
let scope_key = topic_id.map(|t| t.to_string()).unwrap_or_else(|| session_id.clone());
// 转换为 TodoRecord 并持久化
let records: Vec<crate::storage::TodoRecord> = todos_array
.iter()
.filter_map(|item| {
Some(crate::storage::TodoRecord {
id: item.get("id")?.as_str()?.to_string(),
scope_key: scope_key.clone(),
session_id: session_id.clone(),
topic_id: topic_id.map(|t| t.to_string()),
content: item.get("content")?.as_str()?.to_string(),
status: item.get("status")?.as_str()?.to_string(),
priority: item.get("priority")?.as_str()?.to_string(),
created_at: item.get("created_at")?.as_i64()?,
updated_at: item.get("updated_at")?.as_i64()?,
})
})
.collect();
// 持久化到 SQLite
if let Err(e) = self.store.replace_todos(&scope_key, &records) {
tracing::warn!(error = %e, scope_key = %scope_key, "Failed to persist todo list");
}
// 推送到前端（使用 try_send 避免异步）
let summaries: Vec<crate::protocol::TodoItemSummary> = records
.iter()
.map(|r| crate::protocol::TodoItemSummary {
id: r.id.clone(),
content: r.content.clone(),
status: r.status.clone(),
priority: r.priority.clone(),
created_at: r.created_at,
updated_at: r.updated_at,
})
.collect();
let _ = self.user_tx.try_send(crate::protocol::WsOutbound::TodoList {
todos: summaries,
scope_key: scope_key.clone(),
});
break; // 只处理第一个成功的 todo_write
}
}
/// 获取 provider_config 引用
pub fn provider_config(&self) -> &LLMProviderConfig {
&self.provider_config

62
src/gateway/todo_prompt_provider.rs Normal file
View File

@ -0,0 +1,62 @@
use crate::agent::{SystemPrompt, SystemPromptContext, SystemPromptProvider};
pub struct TodoPromptProvider;
impl TodoPromptProvider {
pub fn new() -> Self {
Self
}
}
impl SystemPromptProvider for TodoPromptProvider {
fn build(&self, _context: &SystemPromptContext) -> Option<SystemPrompt> {
Some(SystemPrompt {
content: TODO_WRITE_INSTRUCTIONS.to_string(),
context: Some("todo_write".to_string()),
})
}
}
const TODO_WRITE_INSTRUCTIONS: &str = r#"
## TodoWrite 工具
你可以使用 `todo_write` 工具在对话中维护结构化的任务列表。
### 何时使用
- 当任务有 3 个或以上明确步骤时，应该使用 todo_write 追踪进度
- 不需要为简单的单步操作（如回答一个问题、读取一个文件）创建 todo
### merge 参数
- `merge: false`（默认）：全量替换 — 只传入需要追踪的 todo，不在列表中的项将被移除
- `merge: true`（推荐）：增量更新 — 只传入需要添加或更新的项，未提及的项保持不变。**绝大多数情况应该使用 merge=true，这样你不需要记住所有 id**
### 状态语义
- `pending` — 尚未开始
- `in_progress` — 当前正在执行（同一时间只能有一个）
- `completed` — 已完成
- `cancelled` — 不再需要
### 核心规则
1. 同一时间只能有一个任务处于 `in_progress` 状态
2. 必须先完成当前 `in_progress` 的任务，再开始下一个
3. `completed` 和 `cancelled` 是终端状态，已完成的项不能被重新激活
4. 不要先标记 completed 再去实际执行 — 先完成工作，再标记
5. `content` 字段保持简洁、可执行
### 使用范例
开始任务时（merge 模式）：
```json
{"merge": true, "todos": [{"content": "修复登录 bug", "status": "in_progress"}]}
```
发现新任务时：
```json
{"merge": true, "todos": [{"content": "补充测试", "status": "pending"}]}
```
完成任务时（传入 id + 新状态）：
```json
{"merge": true, "todos": [{"id": "xxx", "content": "修复登录 bug", "status": "completed"}]}
```
"#;

29
src/gateway/tool_registry_factory.rs
View File

@ -1,16 +1,19 @@
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::config::TaskConfig;
use crate::mcp::McpClientManager;
use crate::skills::SkillRuntime;
use crate::storage::{MemoryRepository, SchedulerJobRepository, SkillEventRepository};
use crate::tools::todo_write::TodoItem;
use crate::tools::{
BashTool, CalculatorTool, FileEditTool, FileReadTool, FileWriteTool,
HttpRequestTool, MemoryManageTool, MemorySearchTool,
SchedulerManageTool, SessionMessageSender, SessionSendTool, SkillActivateTool,
SkillManageTool, SubAgentRuntime, TaskTool, TimeTool,
ToolRegistry, WebFetchTool,
TodoWriteTool, ToolRegistry, WebFetchTool,
};
pub(crate) struct ToolRegistryFactory {
@ -25,6 +28,7 @@ pub(crate) struct ToolRegistryFactory {
task_config: TaskConfig,
subagent_runtime: Option<Arc<dyn SubAgentRuntime>>,
mcp_manager: Option<Arc<McpClientManager>>,
todo_state: Option<Arc<RwLock<HashMap<String, Vec<TodoItem>>>>>,
}
impl ToolRegistryFactory {
@ -51,9 +55,18 @@ impl ToolRegistryFactory {
task_config,
subagent_runtime: None,
mcp_manager: None,
todo_state: None,
}
}
pub(crate) fn with_todo_state(
mut self,
state: Arc<RwLock<HashMap<String, Vec<TodoItem>>>>,
) -> Self {
self.todo_state = Some(state);
self
}
pub(crate) fn with_subagent_runtime(
mut self,
runtime: Arc<dyn SubAgentRuntime>,
@ -98,6 +111,11 @@ impl ToolRegistryFactory {
if self.is_enabled("memory_manage") {
registry.register(MemoryManageTool::new(self.memories.clone()));
}
if self.is_enabled("todo_write") {
if let Some(ref state) = self.todo_state {
registry.register(TodoWriteTool::new(state.clone()));
}
}
if self.is_enabled("session_send") {
registry.register(SessionSendTool::new(self.session_message_sender.clone()));
}
@ -198,6 +216,13 @@ impl ToolRegistryFactory {
registry.register(SessionSendTool::new(self.session_message_sender.clone()));
}
// Todo 追踪工具
if self.is_enabled("todo_write") {
if let Some(ref state) = self.todo_state {
registry.register(TodoWriteTool::new(state.clone()));
}
}
// 注册 MCP 工具（如果提供）
if let Some(mcp_tools) = mcp_tools {
for tool in mcp_tools {

16
src/protocol/mod.rs
View File

@ -82,6 +82,17 @@ pub struct SkillSummary {
pub source: String,
}
/// Todo item 摘要（发送给前端）
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TodoItemSummary {
pub id: String,
pub content: String,
pub status: String,
pub priority: String,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchedulerJobSummary {
pub id: String,
@ -257,6 +268,11 @@ pub enum WsOutbound {
},
#[serde(rename = "execution_cancelled")]
ExecutionCancelled { message: String },
#[serde(rename = "todo_list")]
TodoList {
todos: Vec<TodoItemSummary>,
scope_key: String,
},
#[serde(rename = "pong")]
Pong,
}

109
src/storage/mod.rs
View File

@ -14,13 +14,13 @@ pub mod records;
pub use error::StorageError;
pub use ports::{
ConversationRepository, MemoryRepository, PromptInjectionRepository, SchedulerJobRepository,
SkillEventRepository,
SkillEventRepository, TodoRepository,
};
pub use records::{
allowed_namespace_names, get_namespace_description, is_valid_namespace,
ALLOWED_MEMORY_NAMESPACES, GLOBAL_SCOPE_KEY, MemoryRecord, MemoryUpsert, SchedulerJobRecord,
SchedulerJobState, SchedulerJobStatus, SchedulerJobUpsert, SessionRecord, SkillEventRecord,
TopicRecord,
TodoRecord, TopicRecord,
};
#[derive(Clone)]
@ -217,6 +217,7 @@ impl SessionStore {
ensure_messages_schema(&conn)?;
ensure_scheduler_schema(&conn)?;
ensure_memory_scope_key_migration(&conn)?;
ensure_todos_schema(&conn)?;
drop(conn);
@ -1491,6 +1492,74 @@ impl SessionStore {
)
.map_err(StorageError::from)
}
pub fn replace_todos(
&self,
scope_key: &str,
items: &[TodoRecord],
) -> Result<Vec<TodoRecord>, StorageError> {
let conn = self.pool.get()?;
let now = current_timestamp();
// Delete existing todos for this scope_key
conn.execute(
"DELETE FROM todos WHERE scope_key = ?1",
params![scope_key],
)?;
// Insert new todos
for item in items {
conn.execute(
"INSERT INTO todos (id, scope_key, session_id, topic_id, content, status, priority, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
item.id,
scope_key,
item.session_id,
item.topic_id,
item.content,
item.status,
item.priority,
item.created_at,
now,
],
)?;
}
drop(conn);
self.list_todos(scope_key)
}
pub fn list_todos(&self, scope_key: &str) -> Result<Vec<TodoRecord>, StorageError> {
let conn = self.pool.get()?;
let mut stmt = conn.prepare(
"SELECT id, scope_key, session_id, topic_id, content, status, priority, created_at, updated_at
FROM todos
WHERE scope_key = ?1
ORDER BY created_at ASC",
)?;
let rows = stmt.query_map(params![scope_key], |row| {
Ok(TodoRecord {
id: row.get(0)?,
scope_key: row.get(1)?,
session_id: row.get(2)?,
topic_id: row.get(3)?,
content: row.get(4)?,
status: row.get(5)?,
priority: row.get(6)?,
created_at: row.get(7)?,
updated_at: row.get(8)?,
})
})?;
let mut todos = Vec::new();
for row in rows {
todos.push(row?);
}
Ok(todos)
}
}
pub fn persistent_session_id(channel_name: &str, chat_id: &str) -> String {
@ -1800,6 +1869,42 @@ fn ensure_memory_scope_key_migration(conn: &Connection) -> Result<(), StorageErr
Ok(())
}
fn ensure_todos_schema(conn: &Connection) -> Result<(), StorageError> {
let table_exists: bool = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='todos'",
[],
|row| row.get::<_, i64>(0),
)
.map(|count| count > 0)?;
if !table_exists {
conn.execute_batch(
"
CREATE TABLE IF NOT EXISTS todos (
id TEXT PRIMARY KEY,
scope_key TEXT NOT NULL,
session_id TEXT NOT NULL,
topic_id TEXT,
content TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
priority TEXT NOT NULL DEFAULT 'medium',
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_todos_scope
ON todos(scope_key, created_at ASC);
CREATE INDEX IF NOT EXISTS idx_todos_session
ON todos(session_id);
",
)?;
}
Ok(())
}
fn has_column(
conn: &Connection,
table_name: &str,

28
src/storage/ports.rs
View File

@ -1,6 +1,6 @@
use super::{
MemoryRecord, MemoryUpsert, SchedulerJobRecord, SchedulerJobState, SchedulerJobStatus,
SchedulerJobUpsert, SessionRecord, SkillEventRecord, StorageError,
SchedulerJobUpsert, SessionRecord, SkillEventRecord, StorageError, TodoRecord,
};
use crate::bus::ChatMessage;
@ -145,6 +145,18 @@ pub trait SkillEventRepository: Send + Sync + 'static {
) -> Result<Vec<SkillEventRecord>, StorageError>;
}
pub trait TodoRepository: Send + Sync + 'static {
/// Replace all todos for a scope (full replacement pattern).
fn replace_todos(
&self,
scope_key: &str,
todo_records: &[TodoRecord],
) -> Result<Vec<TodoRecord>, StorageError>;
/// Load all todos for a scope, ordered by created_at.
fn list_todos(&self, scope_key: &str) -> Result<Vec<TodoRecord>, StorageError>;
}
impl ConversationRepository for super::SessionStore {
fn ensure_channel_session(
&self,
@ -356,3 +368,17 @@ impl SkillEventRepository for super::SessionStore {
super::SessionStore::list_skill_events(self, session_id)
}
}
impl TodoRepository for super::SessionStore {
fn replace_todos(
&self,
scope_key: &str,
todo_records: &[TodoRecord],
) -> Result<Vec<TodoRecord>, StorageError> {
super::SessionStore::replace_todos(self, scope_key, todo_records)
}
fn list_todos(&self, scope_key: &str) -> Result<Vec<TodoRecord>, StorageError> {
super::SessionStore::list_todos(self, scope_key)
}
}

13
src/storage/records.rs
View File

@ -35,6 +35,19 @@ pub fn allowed_namespace_names() -> Vec<&'static str> {
ALLOWED_MEMORY_NAMESPACES.iter().map(|(name, _)| *name).collect()
}
#[derive(Debug, Clone)]
pub struct TodoRecord {
pub id: String,
pub scope_key: String,
pub session_id: String,
pub topic_id: Option<String>,
pub content: String,
pub status: String,
pub priority: String,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SkillEventRecord {
pub id: String,

2
src/tools/mod.rs
View File

@ -14,6 +14,7 @@ pub mod skill_activate;
pub mod skill_manage;
pub mod task;
pub mod time;
pub mod todo_write;
pub mod traits;
pub mod web_fetch;
@ -39,6 +40,7 @@ pub use task::{
SubagentCatalog, TaskError, TaskRepository, TaskTool,
};
pub use time::TimeTool;
pub use todo_write::TodoWriteTool;
pub use traits::{Tool, ToolContext, ToolResult};
pub use web_fetch::WebFetchTool;

2
src/tools/task/prompt.rs
View File

@ -52,6 +52,8 @@ impl SubagentPromptBuilder {
2. 使用可用的工具进行必要操作\n\
3. 完成后给出简洁的总结\n\
4. 不要尝试创建新的子代理任务\n\n\
任务追踪:\n\
你可以使用 `todo_write` 工具追踪子任务进度。规则：同一时间只有一个 in_progress，完成后再标记下一个，3步以上才使用。\n\n\
注意: 你没有访问主对话历史的权限，这是一个独立的执行上下文。"
} else {
&def.prompt_template

2
src/tools/task/types.rs
View File

@ -61,7 +61,7 @@ impl SubagentDef {
Self {
name: "general".to_string(),
description: "通用型子代理 - 处理复杂多步骤任务".to_string(),
prompt_template: "你是一个专注的子代理，正在执行一个独立任务。\n\n任务描述: {{description}}\n\n你应该:\n1. 专注于完成任务，不要偏离目标\n2. 使用可用的工具进行必要操作\n3. 完成后给出简洁的总结\n4. 不要尝试创建新的子代理任务\n\n注意: 你没有访问主对话历史的权限，这是一个独立的执行上下文。".to_string(),
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,

1153
src/tools/todo_write.rs Normal file
View File

File diff suppressed because it is too large Load Diff