From 351af8870de2d0a23411a6de06c3d84b39d457c4 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Wed, 15 Jul 2026 17:48:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E6=AD=A5=E5=AE=9E=E7=8E=B0todo?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 2 + README.md | 6 + docs/ARCHITECTURE.md | 9 +- .../about-picobot/references/architecture.md | 6 + .../about-picobot/references/db-schema.md | 6 + .../skills/about-picobot/references/tools.md | 5 + src/agent/sub_agent.rs | 147 +++- src/agent/system_prompt.rs | 36 +- src/channels/cli_chat.rs | 48 ++ src/client/mod.rs | 3 + src/gateway/mod.rs | 21 + src/lib.rs | 1 + src/protocol.rs | 14 + src/session/commands.rs | 2 + src/session/events.rs | 5 + src/session/session.rs | 116 ++- src/storage/mod.rs | 70 +- src/tools/delegate.rs | 25 +- src/tools/mod.rs | 4 + src/tools/todo.rs | 163 +++++ src/work/mod.rs | 689 ++++++++++++++++++ tests/test_request_format.rs | 53 ++ webui/src/pages/ChatPage.svelte | 82 ++- webui/src/styles.css | 32 +- 24 files changed, 1502 insertions(+), 43 deletions(-) create mode 100644 src/tools/todo.rs create mode 100644 src/work/mod.rs diff --git a/AGENTS.md b/AGENTS.md index 3db28bc..edd1b7f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -65,6 +65,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del | `skills` | Skills loading, management, and prompt building | `SkillsLoader`, `Skill` | | `storage` | SQLite persistence for sessions and messages | `Storage`, `SessionMeta`, `MessageMeta` | | `scheduler` | Cron-based job scheduling, next-run computation | `Scheduler`, `Schedule`, `next_run_for_schedule()` | +| `work` | Session-scoped active plan and concurrently executable checklist items | `WorkManager`, `TaskPlan`, `TaskItem` | | `observability` | Observer pattern for agent/tool telemetry events | `Observer` trait, `ObserverEvent`, `MultiObserver` | | `protocol` | WebSocket protocol message types | `WsInbound`, `WsOutbound`, `SessionSummary`, `HistoryMessage` | | `config` | Config loading, env substitution, path resolution | `Config`, `LLMProviderConfig` | @@ -76,6 +77,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **Channels** only send/receive messages via `MessageBus`; they know nothing about sessions or LLM - **MessageBus** owns bounded inbound/outbound/control queues; outbound routing and retries belong to `OutboundDispatcher`, not the queue - **SessionManager** owns session state, dialog operations, context construction, per-session work queues, and persistence coordination +- **WorkManager** owns the single active plan per session, item state transitions, plan versions, and plan-change events; plans are optional and absent from ordinary chat context - **Scheduler** supports legacy direct-delivery jobs and managed `task`/`monitor` jobs; managed agents cannot call `send_message`, and `on_alert` suppresses only healthy informational results - **AgentLoop** is stateless across turns; it receives prepared history, calls LLM providers, executes tools, and returns one result - **WebUI management APIs** only expose allowlisted config/profile/log/storage operations; keep response limits, secret redaction, atomic config writes, and profile path allowlists intact diff --git a/README.md b/README.md index 8bbb28b..a5f4135 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ PicoBot 是一个用 Rust 编写的个人 AI 助手运行时。它在本地启 - 在终端里和本地 AI 助手持续对话。 - 在浏览器中聊天，并查看日志、任务和记忆，修改运行配置与助手档案。 +- 复杂任务可创建 session 级 Todo 计划，把不同子项并行委托给多个子 Agent；聊天页侧栏实时显示进度。 - 将同一套 Agent 能力接入飞书/Lark。 - 让 Agent 使用本地文件、Shell、搜索、HTTP、浏览器、MCP 工具完成任务。 - 把长期偏好、事实和历史摘要存成可检索记忆。 @@ -104,6 +105,7 @@ WebUI 随二进制嵌入，不需要 Node.js、npm 或单独部署静态文件 - 在线聊天、会话创建/切换、历史回放，以及基于 Gateway 实时命令清单的 `/` 斜杠命令补全。 - Cron 定时任务、最近运行记录和后台子任务状态。 +- 当前聊天 session 的可展开 Todo 侧栏；计划变化时自动展开，其他 session 的变化显示未读提示。 - Knowledge/Timeline 记忆的分类与全文检索。 - 本地滚动日志的尾部查看、过滤和自动刷新。 - `config.json`、`~/.picobot/USER.md`、`~/.picobot/AGENTS.md` 编辑。 @@ -180,6 +182,7 @@ TUI 会把一个随机客户端标识保存到 `~/.picobot/tui_client_id`，因 | `tools` | Agent 可调用工具集合 | | `storage` | SQLite schema、CRUD、消息和任务持久化 | | `scheduler` | 领取定时任务，执行普通/巡检 Agent，并按投递策略记录或发送结果 | +| `work` | 管理 session 级单 active plan、并行子项状态和 WebSocket 变更事件 | | `skills` | 加载 Skill，并把 Skill 指南注入系统提示 | | `mcp` | 连接 MCP Server，将远端工具包装成普通 Tool | | `task_supervisor` | 统一管理 Gateway 后台任务的取消和有界关停 | @@ -217,6 +220,7 @@ Session ID 使用三段式： | `/dump` | 导出当前 dialog 为 Markdown | | `/mcp` | 查看 MCP 服务器和工具状态 | | `/stop` | 停止当前任务并清空队列 | +| `/todo [done\|cancel]` | 查看、完成或取消当前 session 的任务计划 | | `/?`, `/help` | 查看帮助 | ### 记忆 @@ -244,6 +248,7 @@ PicoBot 有两类记忆： | `get_skill` | 列出或读取本地 Skill | | `memory_store` / `memory_recall` / `timeline_recall` / `memory_forget` | 长期记忆操作 | | `delegate` | 启动 inline、background 或 parallel 子 Agent | +| `todo` | 为复杂、多轮任务创建并更新当前 session 的持久化计划 | | `send_message` | 向指定渠道发送消息 | | `chat_manager` | 查看渠道、会话和历史消息 | | `cron_add/list/remove/enable/disable/update` | 管理定时任务 | @@ -354,6 +359,7 @@ src/ observability/ Agent/tool telemetry observer providers/ OpenAI 兼容和 Anthropic provider scheduler/ 定时任务运行时 + work/ Session 级任务计划与并行子项状态机 session/ 会话生命周期、dialog 命令、持久化集成 skills/ Skill 加载和内置 Skill 安装 storage/ SQLite schema 和 CRUD diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e5027fe..e91b502 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -65,6 +65,7 @@ flowchart LR | `storage` | SQLite schema、迁移、原子 CRUD | 运行时调度策略 | | `memory` | Knowledge/Timeline 的存取与召回 | 直接驱动消息发送 | | `scheduler` | 领取到期任务、执行普通/巡检 Agent、应用投递策略、原子记录结果 | 复用聊天会话历史、直接感知 Channel | +| `work` | session 级单 active plan、并行子项状态机、版本和变更事件 | 执行模型调用、持有 Channel/WebSocket | | `task_supervisor` | 后台任务注册、取消、限时回收 | 业务级重试和结果语义 | ## 4. 消息与控制数据流 @@ -148,7 +149,9 @@ Session ID 格式为： 5. 持久化写入由 `persistence_lock` 串行化；多条相关记录应使用 Storage 的原子接口。 6. 内存先变更但持久化失败时，必须回滚精确匹配的消息后缀，不能删除无关的新状态。 -SessionManager 负责组装会话上下文：系统提示、Skills、召回的 Knowledge、压缩后的 Timeline 和当前消息历史。`AgentLoop` 接收完整输入执行一次模型/工具循环，本身不拥有会话状态。 +SessionManager 负责组装会话上下文：系统提示、Skills、召回的 Knowledge、压缩后的 Timeline、可选的 active plan 摘要和当前消息历史。普通闲聊 session 没有 plan 摘要；计划状态由 `WorkManager` 从 SQLite 读取，在历史压缩之后追加，因此不以自然语言摘要作为权威来源。`AgentLoop` 接收完整输入执行一次模型/工具循环，本身不拥有会话状态。 + +每个 session 最多有一个 active plan，但 plan 内多个 item 可以分别绑定不同子 Agent 并行执行。主 Agent 通过 `todo` 创建和维护计划，通过 `delegate.plan_item_id` 委托；子 Agent 不获得 `todo` 或 `delegate`。领取 item 使用条件更新防止重复执行，迟到结果只有在 plan 仍 active 且 execution ID 匹配时才允许提交。 ## 6. 持久化 @@ -159,7 +162,7 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K - 5 秒 busy timeout。 - schema version 迁移。 -持久化范围包括 sessions、messages、memories、scheduled jobs、job runs 和 background tasks。修改 schema 时应： +持久化范围包括 sessions、messages、memories、task plans/items、scheduled jobs、job runs 和 background tasks。修改 schema 时应： 1. 更新集中式 schema/迁移逻辑。 2. 保留已有数据库的升级路径。 @@ -194,7 +197,7 @@ WebSocket 每个客户端的 writer task 是连接局部任务，由连接 handl ### WebUI 与管理 API -Gateway 在 `/` 提供随二进制编译的 HTML/CSS/JavaScript，不依赖外部 CDN 或前端运行服务。前端源码位于 `webui/`，由 Svelte 5 + Vite 构建，Bits UI 提供无样式的可访问组件原语。`build.rs` 监听前端源码、锁文件和构建配置，增量地将生产资源生成到 Cargo `OUT_DIR`，Rust 再从该目录编译嵌入；前端产物不进入仓库，最终用户使用发布二进制时不需要 Node.js。浏览器聊天继续使用 `/ws` 和 `cli_chat` 渠道，因此复用现有 dialog scope、每会话串行 worker、历史持久化和出站 lane；斜杠命令补全通过 `get_slash_commands` 获取 Gateway 的实时命令与别名清单，不在前端重复定义；WebUI 不直接调用 Provider 或 SessionManager。 +Gateway 在 `/` 提供随二进制编译的 HTML/CSS/JavaScript，不依赖外部 CDN 或前端运行服务。前端源码位于 `webui/`，由 Svelte 5 + Vite 构建，Bits UI 提供无样式的可访问组件原语。`build.rs` 监听前端源码、锁文件和构建配置，增量地将生产资源生成到 Cargo `OUT_DIR`，Rust 再从该目录编译嵌入；前端产物不进入仓库，最终用户使用发布二进制时不需要 Node.js。浏览器聊天继续使用 `/ws` 和 `cli_chat` 渠道，因此复用现有 dialog scope、每会话串行 worker、历史持久化和出站 lane；聊天页的 Todo 侧栏默认隐藏，按 session 保存快照和未读状态，并通过结构化 `session_plan`/`plan_updated` 帧刷新，计划变化不会写入聊天历史；斜杠命令补全通过 `get_slash_commands` 获取 Gateway 的实时命令与别名清单，不在前端重复定义；WebUI 不直接调用 Provider 或 SessionManager。 同源 `/api/*` 管理接口只提供显式白名单能力： diff --git a/resources/skills/about-picobot/references/architecture.md b/resources/skills/about-picobot/references/architecture.md index d86ef48..6d407b2 100644 --- a/resources/skills/about-picobot/references/architecture.md +++ b/resources/skills/about-picobot/references/architecture.md @@ -32,6 +32,7 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message | `memory` | 长期记忆存储与检索 | | `mcp` | MCP（Model Context Protocol）工具集成 | | `task_supervisor` | Gateway 后台任务注册、取消、限时等待和强制回收 | +| `work` | 每个 session 的单 active plan、并行子项状态、版本与变更事件 | ## 功能边界 @@ -43,6 +44,7 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message - Tools 接收原始参数，返回字符串结果 - MCP 工具在 Gateway 初始化时连接服务器、发现工具，并包装成普通 Tool 注册到 ToolRegistry - 子 Agent 由 `delegate` 工具创建，复用 provider 配置和按需过滤后的工具集；后台任务结果通过 MessageBus 发回原会话 +- 复杂任务可使用 `todo` 创建 session 级计划；多个子项可通过 `delegate.plan_item_id` 并行委托，子 Agent 不能修改计划 - WebUI 聊天复用 `/ws` 与 `cli_chat`；同源管理 API 只读取受限的日志、任务、记忆，并对白名单配置文件做原子写入 - WebUI 使用 Svelte 5 + Vite，Bits UI 提供无样式可访问组件；`cargo build` 增量生成前端到 Cargo `OUT_DIR`，再嵌入单二进制，仓库不保存生成产物 @@ -243,6 +245,10 @@ Gateway 初始化时读取 `config.mcp.servers`： 后台子 Agent 通过 `TaskSupervisor::spawn_graceful` 注册，受 `gateway.max_concurrent_background_tasks` 限制；Gateway 关停时先收到取消信号，再在总宽限期内清理。 +## Session Todo 计划 + +每个 session 最多有一个 active plan；普通闲聊没有计划上下文。计划和子项分别持久化到 `task_plans`、`task_items`，历史压缩后仍从权威状态生成精简摘要。WebUI 聊天页通过 `session_plan` 和 `plan_updated` WebSocket 帧显示默认隐藏的侧栏，当前 session 变化时自动展开，其他 session 只标记未读。 + --- ## 出站投递与关停 diff --git a/resources/skills/about-picobot/references/db-schema.md b/resources/skills/about-picobot/references/db-schema.md index 2893b29..57d6b1f 100644 --- a/resources/skills/about-picobot/references/db-schema.md +++ b/resources/skills/about-picobot/references/db-schema.md @@ -66,6 +66,12 @@ delegate 后台子任务表。`session_id` 不使用数据库外键，因为 ses | `finished_at` | INTEGER | 结束时间 | | `created_at` | INTEGER | 创建时间 | +## task_plans / task_items 表 + +`task_plans` 保存 session 级任务计划，通过部分唯一索引保证每个 session 最多一个 `status='active'` 的计划。`version` 在任何子项变化时递增，用于 WebSocket 快照排序和乐观并发检查。 + +`task_items` 以 `(plan_id, id)` 为复合主键，因此每个计划都可使用 `T1`、`T2` 等稳定显示 ID。子项状态为 `pending`、`in_progress`、`completed` 或 `blocked`；`executor_kind` 和 `execution_id` 将并行子 Agent 执行绑定到具体子项。只有 active plan 且 execution ID 匹配时，异步结果才能提交。 + ## memories 表 长期记忆存储。 diff --git a/resources/skills/about-picobot/references/tools.md b/resources/skills/about-picobot/references/tools.md index 00e28ff..8cef6db 100644 --- a/resources/skills/about-picobot/references/tools.md +++ b/resources/skills/about-picobot/references/tools.md @@ -143,9 +143,14 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 | `timeout_secs` | 否 | 超时秒数，默认 3600 | | `tasks` | parallel 必填 | 并行子任务数组 | | `task_id` | 查询/取消必填 | 后台任务 ID | +| `plan_item_id` | 否 | 将 inline/background 子 Agent 绑定到当前计划子项；parallel 数组中的每项也可分别绑定 | 默认只读工具集：`file_read`、`file_search`、`content_search`、`web_fetch`、`http_request`、`calculator`。 +## todo — Session 任务计划 + +仅用于明确的复杂、多轮或并行任务。`create` 创建当前 session 唯一的 active plan；`view` 查看；`append` 增加子项；`update` 修改子项状态；`close` 完成或取消计划。普通闲聊和单步操作不应创建计划。子 Agent 始终被过滤掉 `todo` 和 `delegate`，计划结构只由主 Agent 管理。 + --- ## browser — 浏览器自动化 diff --git a/src/agent/sub_agent.rs b/src/agent/sub_agent.rs index 0404179..086d4b2 100644 --- a/src/agent/sub_agent.rs +++ b/src/agent/sub_agent.rs @@ -47,6 +47,8 @@ pub struct SubAgentConfig { pub allowed_tools: Option>, pub max_iterations: Option, pub timeout_secs: Option, + pub plan_item_id: Option, + pub session_id: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -122,6 +124,7 @@ pub struct SubAgentManager { notify_tx: tokio::sync::mpsc::UnboundedSender, max_concurrent_background_tasks: usize, skills_loader: Option>, + work_manager: Option>, task_supervisor: crate::task_supervisor::TaskSupervisor, } @@ -144,10 +147,16 @@ impl SubAgentManager { notify_tx, max_concurrent_background_tasks, skills_loader, + work_manager: None, task_supervisor, } } + pub fn with_work_manager(mut self, work_manager: Arc) -> Self { + self.work_manager = Some(work_manager); + self + } + pub fn filter_tools(&self, allowed: &Option>) -> Arc { let allowed_set: HashSet<&str> = match allowed { Some(list) => list.iter().map(|s| s.as_str()).collect(), @@ -155,7 +164,7 @@ impl SubAgentManager { }; let filtered = ToolRegistry::new(); for (name, tool) in self.full_tools.iter() { - if allowed_set.contains(name.as_str()) && name != "delegate" { + if allowed_set.contains(name.as_str()) && name != "delegate" && name != "todo" { filtered.register_raw(name, tool); } } @@ -230,6 +239,7 @@ impl SubAgentManager { let agent = self .build_sub_agent(&config, tools) .map_err(|e| SubAgentError::ProviderCreation(e.to_string()))?; + self.assign_work_item(&config, &task_id).await?; let history = vec![ ChatMessage::system(system_prompt), @@ -246,7 +256,7 @@ impl SubAgentManager { let duration_ms = start.elapsed().as_millis() as u64; - match result { + let result = match result { Ok(Ok(agent_result)) => { let (content, truncated) = truncate_sub_agent_result(&agent_result.final_response.content); @@ -260,35 +270,37 @@ impl SubAgentManager { .iter() .filter(|m| m.role == "assistant" && m.tool_calls.is_some()) .count(); - Ok(SubAgentResult { - task_id, + SubAgentResult { + task_id: task_id.clone(), content, content_truncated: truncated, status: TaskStatus::Completed, tool_calls_count, iterations, duration_ms, - }) + } } - Ok(Err(e)) => Ok(SubAgentResult { - task_id, + Ok(Err(e)) => SubAgentResult { + task_id: task_id.clone(), content: String::new(), content_truncated: false, status: TaskStatus::Failed(e.to_string()), tool_calls_count: 0, iterations: 0, duration_ms, - }), - Err(_elapsed) => Ok(SubAgentResult { - task_id, + }, + Err(_elapsed) => SubAgentResult { + task_id: task_id.clone(), content: String::new(), content_truncated: false, status: TaskStatus::TimedOut, tool_calls_count: 0, iterations: 0, duration_ms, - }), - } + }, + }; + self.finish_work_item(&config, &result).await; + Ok(result) } pub async fn run_parallel( @@ -319,11 +331,12 @@ impl SubAgentManager { .map_err(|_| SubAgentError::TooManyTasks(self.max_concurrent_background_tasks))?; let task_id = generate_task_id(); + let mut work_config = config.clone(); + if work_config.session_id.is_none() { + work_config.session_id = Some(ctx.session_id.clone()); + } let cancel_token = CancellationToken::new(); - self.active_tasks - .insert(task_id.clone(), cancel_token.clone()); - // Write DB: pending if let Some(ref storage) = self.storage { let allowed_tools_json = config @@ -351,6 +364,28 @@ impl SubAgentManager { .await .map_err(|e| SubAgentError::Storage(e.to_string()))?; } + if let Err(error) = self.assign_work_item(&work_config, &task_id).await { + if let Some(ref storage) = self.storage { + let _ = storage + .update_background_task_status( + &task_id, + crate::storage::background_task::BackgroundTaskUpdate { + status: "cancelled", + result: None, + error: Some("plan item assignment failed"), + started_at: None, + finished_at: Some(chrono::Utc::now().timestamp_millis()), + tool_calls_count: None, + iterations: None, + }, + ) + .await; + } + return Err(error); + } + + self.active_tasks + .insert(task_id.clone(), cancel_token.clone()); let tools = self.filter_tools(&config.allowed_tools); let timeout_secs = config.timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS); @@ -381,6 +416,9 @@ impl SubAgentManager { let ch = ctx.channel.clone(); let cid = ctx.chat_id.clone(); let prompt = config.prompt.clone(); + let work_manager = self.work_manager.clone(); + let work_item_id = work_config.plan_item_id.clone(); + let work_session_id = work_config.session_id.clone(); let spawned = self.task_supervisor.spawn_graceful( format!("sub-agent:{task_id}"), @@ -510,6 +548,23 @@ impl SubAgentManager { TaskStatus::TimedOut => ("failed".to_string(), Some("timeout".to_string())), }; + if let (Some(manager), Some(session_id), Some(item_id)) = + (work_manager.as_ref(), work_session_id.as_deref(), work_item_id.as_deref()) + { + let completed = matches!(result.status, TaskStatus::Completed); + let summary = if completed { + Some(result.content.as_str()) + } else { + error_val.as_deref().or(Some("子 Agent 未完成任务")) + }; + if let Err(error) = manager + .finish_sub_agent(session_id, item_id, &tid, completed, summary) + .await + { + tracing::warn!(task_id = %tid, item_id, error = %error, "Failed to update plan item"); + } + } + if let Some(ref s) = storage { let _ = s .update_background_task_status(&tid, crate::storage::background_task::BackgroundTaskUpdate { @@ -554,6 +609,21 @@ impl SubAgentManager { ) .await; } + if let (Some(manager), Some(session_id), Some(item_id)) = ( + self.work_manager.as_ref(), + work_config.session_id.as_deref(), + work_config.plan_item_id.as_deref(), + ) { + let _ = manager + .finish_sub_agent( + session_id, + item_id, + &task_id, + false, + Some("gateway shutdown"), + ) + .await; + } return Err(SubAgentError::Other( "gateway is shutting down and cannot accept background tasks".to_string(), )); @@ -632,6 +702,51 @@ impl SubAgentManager { pub fn active_task_count(&self) -> usize { self.active_tasks.len() } + + async fn assign_work_item( + &self, + config: &SubAgentConfig, + task_id: &str, + ) -> Result<(), SubAgentError> { + if let (Some(manager), Some(session_id), Some(item_id)) = ( + self.work_manager.as_ref(), + config.session_id.as_deref(), + config.plan_item_id.as_deref(), + ) { + manager + .assign_sub_agent(session_id, item_id, task_id) + .await + .map_err(|error| SubAgentError::Storage(error.to_string()))?; + } + Ok(()) + } + + async fn finish_work_item(&self, config: &SubAgentConfig, result: &SubAgentResult) { + let (Some(manager), Some(session_id), Some(item_id)) = ( + self.work_manager.as_ref(), + config.session_id.as_deref(), + config.plan_item_id.as_deref(), + ) else { + return; + }; + let completed = matches!(result.status, TaskStatus::Completed); + let summary = if completed { + Some(result.content.as_str()) + } else { + match &result.status { + TaskStatus::Failed(error) => Some(error.as_str()), + TaskStatus::TimedOut => Some("子 Agent 执行超时"), + TaskStatus::Cancelled => Some("子 Agent 已取消"), + TaskStatus::Completed => None, + } + }; + if let Err(error) = manager + .finish_sub_agent(session_id, item_id, &result.task_id, completed, summary) + .await + { + tracing::warn!(task_id = %result.task_id, item_id, error = %error, "Failed to update plan item"); + } + } } fn generate_task_id() -> String { @@ -722,6 +837,8 @@ mod tests { allowed_tools: None, max_iterations: None, timeout_secs: Some(1), + plan_item_id: None, + session_id: None, }, DelegateContext { session_id: "cli:test:dialog".into(), diff --git a/src/agent/system_prompt.rs b/src/agent/system_prompt.rs index 548b986..8231240 100644 --- a/src/agent/system_prompt.rs +++ b/src/agent/system_prompt.rs @@ -46,6 +46,7 @@ impl SystemPromptBuilder { Box::new(SafetySection), Box::new(CrossChannelSection), Box::new(MemorySection), + Box::new(WorkManagementSection), Box::new(DelegationSection), ], } @@ -319,6 +320,26 @@ impl PromptSection for MemorySection { /// Sub-agent delegation principles. pub struct DelegationSection; +/// Optional session-scoped planning guidance. The actual active plan is +/// injected dynamically only when one exists. +pub struct WorkManagementSection; + +impl PromptSection for WorkManagementSection { + fn name(&self) -> &str { + "work_management" + } + + fn build(&self, _ctx: &PromptContext<'_>) -> String { + "## 任务追踪\n\n\ + - 普通闲聊、问答和单步操作不要创建计划。\n\ + - 用户明确要求规划，或任务需要至少三个可验证步骤、跨多个轮次、后台等待或并行委托时，使用 todo 创建计划。\n\ + - 计划存在时按真实进展更新子项；不要重复执行已分配给子 Agent 的子项。\n\ + - 子 Agent 可以并行执行不同子项；主 Agent 负责计划、整合、验证和关闭计划。\n\ + - 阻塞时记录原因，所有子项完成后才能关闭为 completed。" + .to_string() + } +} + impl PromptSection for DelegationSection { fn name(&self) -> &str { "delegation" @@ -452,7 +473,11 @@ pub fn build_system_prompt(workspace_dir: &Path, model_name: &str, tools: &ToolR } /// Build a runtime context tail that should be appended to the latest user message. -pub fn build_runtime_context(session_id: Option<&str>, memory_context: Option<&str>) -> String { +pub fn build_runtime_context( + session_id: Option<&str>, + memory_context: Option<&str>, + work_context: Option<&str>, +) -> String { let mut sections = Vec::new(); let now = chrono::Local::now(); @@ -470,6 +495,10 @@ pub fn build_runtime_context(session_id: Option<&str>, memory_context: Option<&s sections.push(format!("### 记忆上下文\n\n{}", context)); } + if let Some(context) = work_context.filter(|s| !s.trim().is_empty()) { + sections.push(context.to_string()); + } + if sections.is_empty() { String::new() } else { @@ -574,7 +603,8 @@ mod tests { let _ = (temp_dir, tools); - let prompt = build_runtime_context(Some("session-123"), Some("- user_pref: Prefers Rust")); + let prompt = + build_runtime_context(Some("session-123"), Some("- user_pref: Prefers Rust"), None); assert!(prompt.contains("## 运行时上下文")); assert!(prompt.contains("session-123")); assert!(prompt.contains("Prefers Rust")); @@ -587,7 +617,7 @@ mod tests { let _ = (temp_dir, tools); - let prompt = build_runtime_context(None, None); + let prompt = build_runtime_context(None, None, None); assert!(prompt.contains("## 运行时上下文")); assert!(prompt.contains("当前日期与时间")); } diff --git a/src/channels/cli_chat.rs b/src/channels/cli_chat.rs index 46c5c44..3f061b5 100644 --- a/src/channels/cli_chat.rs +++ b/src/channels/cli_chat.rs @@ -98,6 +98,29 @@ impl CliChatChannel { } } + /// Push a structured task-plan update to the WebSocket client that owns + /// the session. Other channels remain unaffected. + pub async fn publish_plan_changed(&self, event: crate::work::PlanChanged) { + let Some(session_id) = UnifiedSessionId::parse(&event.session_id) else { + return; + }; + if session_id.channel != "cli_chat" { + return; + } + let client = self.clients.lock().await.get(&session_id.chat_id).cloned(); + if let Some(client) = client { + let _ = client + .sender + .send(WsOutbound::PlanUpdated { + session_id: event.session_id, + reason: event.reason, + changed_item_ids: event.changed_item_ids, + plan: event.plan, + }) + .await; + } + } + /// Handle an inbound message from a client pub(crate) async fn handle_inbound(&self, client: Arc, raw_msg: &str) { match parse_inbound(raw_msg) { @@ -390,6 +413,31 @@ impl CliChatChannel { } } } + WsInbound::GetSessionPlan { session_id } => { + let unified_id = Self::parse_client_session(&client, &session_id)?; + let (reply_tx, mut reply_rx) = mpsc::channel(1); + bus.publish_control(ControlMessage { + op: SessionCommand::GetTaskPlan { + session_id: unified_id, + }, + reply_tx, + }) + .await?; + match reply_rx.recv().await { + Some(Ok(SessionEvent::TaskPlan { session_id, plan })) => { + let _ = client + .sender + .send(WsOutbound::SessionPlan { + session_id: session_id.to_string(), + plan, + }) + .await; + } + Some(Ok(_)) => {} + Some(Err(error)) => return Err(error), + None => return Err(ChannelError::Other("Control channel closed".to_string())), + } + } WsInbound::RenameSession { session_id, title } => { let target = session_id .or(current_session_guard.clone()) diff --git a/src/client/mod.rs b/src/client/mod.rs index ffef1cb..251be12 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -194,6 +194,9 @@ async fn handle_ws_message(app: &mut App, outbound: WsOutbound) { session_id, messages, } => app.set_history(&session_id, messages), + // The first Todo UI is WebUI-only. CLI keeps receiving ordinary task + // notifications and may inspect plans through /todo. + WsOutbound::SessionPlan { .. } | WsOutbound::PlanUpdated { .. } => {} WsOutbound::SessionRenamed { session_id, title } => { if let Some(session) = app .sessions diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 7ccf3d0..04cf05b 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -211,6 +211,22 @@ impl GatewayState { let bus_for_outbound = bus.clone(); let session_manager = self.session_manager.clone(); + // Relay structured plan changes to WebSocket clients. This remains + // separate from chat messages, so task UI updates never pollute history. + let mut plan_events = self.session_manager.work_manager().subscribe(); + let cli_chat = self.cli_chat_channel(); + self.task_supervisor.spawn("task-plan-events", async move { + loop { + match plan_events.recv().await { + Ok(event) => cli_chat.publish_plan_changed(event).await, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + tracing::warn!(skipped, "Task plan event relay lagged"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }); + // Spawn unified message processor // This handles both inbound AI messages and control messages in one loop self.task_supervisor.spawn("message-processor", async move { @@ -362,6 +378,11 @@ impl GatewayState { messages, }) .map_err(|e| ChannelError::Other(e.to_string())), + GetTaskPlan { session_id } => session_manager + .get_task_plan(&session_id) + .await + .map(|plan| SessionEvent::TaskPlan { session_id, plan }) + .map_err(|e| ChannelError::Other(e.to_string())), RenameDialog { session_id, title } => session_manager .rename_dialog(&session_id, &title) .await diff --git a/src/lib.rs b/src/lib.rs index 19b844c..3aea9f1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,3 +18,4 @@ pub mod storage; pub mod task_supervisor; pub mod tools; pub mod util; +pub mod work; diff --git a/src/protocol.rs b/src/protocol.rs index c666f3f..3dc6f08 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -66,6 +66,8 @@ pub enum WsInbound { #[serde(default, skip_serializing_if = "Option::is_none")] limit: Option, }, + #[serde(rename = "get_session_plan")] + GetSessionPlan { session_id: String }, #[serde(rename = "rename_session")] RenameSession { #[serde(default, skip_serializing_if = "Option::is_none")] @@ -122,6 +124,18 @@ pub enum WsOutbound { session_id: String, messages: Vec, }, + #[serde(rename = "session_plan")] + SessionPlan { + session_id: String, + plan: Option, + }, + #[serde(rename = "plan_updated")] + PlanUpdated { + session_id: String, + reason: String, + changed_item_ids: Vec, + plan: Option, + }, #[serde(rename = "session_renamed")] SessionRenamed { session_id: String, title: String }, #[serde(rename = "session_archived")] diff --git a/src/session/commands.rs b/src/session/commands.rs index 001062c..24b35db 100644 --- a/src/session/commands.rs +++ b/src/session/commands.rs @@ -26,6 +26,8 @@ pub enum SessionCommand { session_id: UnifiedSessionId, limit: u32, }, + /// Load the active task plan for a dialog. + GetTaskPlan { session_id: UnifiedSessionId }, /// Get the current dialog for a chat GetCurrentDialog { channel: String, chat_id: String }, /// Rename a dialog diff --git a/src/session/events.rs b/src/session/events.rs index 77046c4..ec3a024 100644 --- a/src/session/events.rs +++ b/src/session/events.rs @@ -36,6 +36,11 @@ pub enum SessionEvent { session_id: UnifiedSessionId, messages: Vec, }, + /// Active task plan for a dialog, if present. + TaskPlan { + session_id: UnifiedSessionId, + plan: Option, + }, /// Dialog renamed DialogRenamed { session_id: UnifiedSessionId, diff --git a/src/session/session.rs b/src/session/session.rs index 5c07c05..5ab3d5b 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -120,6 +120,7 @@ struct AgentTask { struct AgentWorkerDeps { bus: Arc, memory_manager: Arc, + work_manager: Arc, skills_loader: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, } @@ -893,6 +894,7 @@ pub struct SessionManager { storage: Arc, pub(super) bus: Arc, memory_manager: Arc, + work_manager: Arc, sub_agent_manager: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, } @@ -982,6 +984,11 @@ pub static SLASH_COMMANDS: &[SlashCommand] = &[ description: "停止当前正在执行的任务并清空消息队列", aliases: &["/stop"], }, + SlashCommand { + name: "todo", + description: "查看、完成或取消当前任务计划", + aliases: &["/todo"], + }, ]; fn resolve_slash_command(command: &str) -> Option<&'static SlashCommand> { @@ -1000,6 +1007,7 @@ impl SessionManager { AgentWorkerDeps { bus: self.bus.clone(), memory_manager: self.memory_manager.clone(), + work_manager: self.work_manager.clone(), skills_loader: self.skills_loader.clone(), task_supervisor: self.task_supervisor.clone(), } @@ -1019,24 +1027,29 @@ impl SessionManager { skills_loader.set_workspace_skills_dir(provider_config.workspace_dir.clone()); let skills_loader = Arc::new(skills_loader); + let work_manager = Arc::new(crate::work::WorkManager::new(storage.clone())); let tools = Arc::new(create_default_tools( skills_loader.clone(), memory_manager.clone(), + work_manager.clone(), None, // SubAgentManager created below browser_config.as_ref(), )); // Create SubAgentManager and register DelegateTool let (notify_tx, mut notify_rx) = tokio::sync::mpsc::unbounded_channel(); - let sub_agent_manager = Arc::new(crate::agent::SubAgentManager::new( - provider_config.clone(), - tools.clone(), - Some(storage.clone()), - notify_tx, - max_concurrent_background_tasks, - Some(skills_loader.clone()), - task_supervisor.clone(), - )); + let sub_agent_manager = Arc::new( + crate::agent::SubAgentManager::new( + provider_config.clone(), + tools.clone(), + Some(storage.clone()), + notify_tx, + max_concurrent_background_tasks, + Some(skills_loader.clone()), + task_supervisor.clone(), + ) + .with_work_manager(work_manager.clone()), + ); tools.register(crate::tools::DelegateTool::new(sub_agent_manager.clone())); // Start background task notification consumer @@ -1045,13 +1058,17 @@ impl SessionManager { while let Some(notif) = notify_rx.recv().await { let content = format_task_notification(¬if.task_id, ¬if.status, ¬if.result_summary); + let metadata = HashMap::from([ + ("_type".to_string(), "notification".to_string()), + ("_session_id".to_string(), notif.session_id), + ]); let outbound = OutboundMessage { channel: notif.channel, chat_id: notif.chat_id, content, reply_to: None, media: vec![], - metadata: std::collections::HashMap::new(), + metadata, delivery: None, }; let _ = sm_bus.publish_outbound(outbound).await; @@ -1088,6 +1105,7 @@ impl SessionManager { storage, bus, memory_manager, + work_manager, sub_agent_manager, task_supervisor, }) @@ -1104,6 +1122,10 @@ impl SessionManager { self.tools.clone() } + pub fn work_manager(&self) -> Arc { + self.work_manager.clone() + } + /// 为定时任务创建一个无 session 绑定的 AgentLoop pub fn create_cron_agent(&self) -> Result { let provider = create_provider(self.provider_config.clone()) @@ -1452,6 +1474,49 @@ impl SessionManager { }; Ok((None, resp)) } + "todo" => { + let sid = current_session_id + .ok_or_else(|| AgentError::Other("no active session".to_string()))?; + let action = args.map(str::trim).filter(|value| !value.is_empty()); + match action { + Some("cancel") => self + .work_manager + .close_plan(&sid.to_string(), "cancelled", None) + .await + .map(|plan| (None, format!("任务计划已取消：{}", plan.objective))) + .map_err(|error| AgentError::Other(error.to_string())), + Some("done") => self + .work_manager + .close_plan(&sid.to_string(), "completed", None) + .await + .map(|plan| (None, format!("任务计划已完成：{}", plan.objective))) + .map_err(|error| AgentError::Other(error.to_string())), + Some(_) => Err(AgentError::Other("Usage: /todo [done|cancel]".to_string())), + None => match self.work_manager.active_plan(&sid.to_string()).await { + Ok(Some(plan)) => { + let mut lines = vec![format!( + "任务计划：{}（version {}）", + plan.objective, plan.version + )]; + for item in plan.items { + let icon = match item.status.as_str() { + "completed" => "✓", + "in_progress" => "●", + "blocked" => "!", + _ => "○", + }; + lines.push(format!( + "{icon} {} [{}] {}", + item.id, item.status, item.title + )); + } + Ok((None, lines.join("\n"))) + } + Ok(None) => Ok((None, "当前 session 没有 active plan。".to_string())), + Err(error) => Err(AgentError::Other(error.to_string())), + }, + } + } _ => Err(AgentError::Other(format!( "未知命令：/{}。输入 /? 获取帮助。", cmd.name @@ -1674,6 +1739,21 @@ impl SessionManager { .map_err(|e| AgentError::Other(format!("failed to load dialog history: {e}"))) } + pub async fn get_task_plan( + &self, + session_id: &UnifiedSessionId, + ) -> Result, AgentError> { + let session_id = session_id.to_string(); + self.storage + .get_session(&session_id) + .await + .map_err(|error| AgentError::Other(format!("failed to load dialog: {error}")))?; + self.work_manager + .plan_for_session(&session_id) + .await + .map_err(|error| AgentError::Other(format!("failed to load task plan: {error}"))) + } + pub async fn list_dialogs( &self, channel: &str, @@ -2084,6 +2164,7 @@ fn spawn_agent_worker( let AgentWorkerDeps { bus, memory_manager, + work_manager, skills_loader, task_supervisor, } = deps; @@ -2224,8 +2305,19 @@ fn spawn_agent_worker( _ => None, }; - let runtime_context = - build_runtime_context(Some(unified_str.as_str()), memory_context.as_deref()); + let work_context = match work_manager.active_plan(&unified_str).await { + Ok(Some(plan)) => Some(plan.compact_context()), + Ok(None) => None, + Err(error) => { + tracing::warn!(error = %error, "Failed to load active task plan"); + None + } + }; + let runtime_context = build_runtime_context( + Some(unified_str.as_str()), + memory_context.as_deref(), + work_context.as_deref(), + ); let system_prompt_out = { let guard = session.lock().await; diff --git a/src/storage/mod.rs b/src/storage/mod.rs index a44fdd2..3628f55 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -14,7 +14,7 @@ use sqlx::{Pool, Row, Sqlite}; use std::path::Path; use tokio::time::{Duration, sleep}; -const SCHEMA_VERSION: i64 = 2; +const SCHEMA_VERSION: i64 = 3; const INSERT_MESSAGE_SQL: &str = r#" INSERT INTO messages (id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) @@ -170,6 +170,64 @@ impl Storage { .execute(&self.pool) .await?; + // Session-scoped task plans. A session may have at most one active plan, + // while independent items can be executed concurrently. + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS task_plans ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL, + objective TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + version INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + closed_at INTEGER + ) + "#, + ) + .execute(&self.pool) + .await?; + + sqlx::query( + r#" + CREATE UNIQUE INDEX IF NOT EXISTS idx_task_plans_active_session + ON task_plans(session_id) WHERE status = 'active' + "#, + ) + .execute(&self.pool) + .await?; + + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS task_items ( + id TEXT NOT NULL, + plan_id TEXT NOT NULL, + ordinal INTEGER NOT NULL, + title TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + executor_kind TEXT, + execution_id TEXT, + result_summary TEXT, + error TEXT, + version INTEGER NOT NULL DEFAULT 1, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY(plan_id, id), + FOREIGN KEY(plan_id) REFERENCES task_plans(id) ON DELETE CASCADE, + UNIQUE(plan_id, ordinal) + ) + "#, + ) + .execute(&self.pool) + .await?; + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_task_items_plan ON task_items(plan_id, ordinal)", + ) + .execute(&self.pool) + .await?; + sqlx::query( r#" CREATE TABLE IF NOT EXISTS memories ( @@ -1556,6 +1614,16 @@ mod tests { .await .unwrap(); assert_eq!(schema_version, SCHEMA_VERSION); + for table in ["task_plans", "task_items"] { + let exists: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?", + ) + .bind(table) + .fetch_one(storage.pool()) + .await + .unwrap(); + assert_eq!(exists, 1, "missing migrated table {table}"); + } } #[tokio::test] diff --git a/src/tools/delegate.rs b/src/tools/delegate.rs index fc8827e..5baea17 100644 --- a/src/tools/delegate.rs +++ b/src/tools/delegate.rs @@ -71,6 +71,10 @@ impl Tool for DelegateTool { "type": "array", "items": { "type": "string" }, "description": "该子任务的工具列表" + }, + "plan_item_id": { + "type": "string", + "description": "可选，绑定当前计划中的子项 ID（如 T2）" } }, "required": ["prompt"] @@ -79,6 +83,10 @@ impl Tool for DelegateTool { "task_id": { "type": "string", "description": "后台任务ID（action=check_task/cancel_task 时必填）" + }, + "plan_item_id": { + "type": "string", + "description": "inline/background 模式可选，绑定当前计划中的子项 ID" } }, "required": ["action"] @@ -133,6 +141,8 @@ impl DelegateTool { allowed_tools, max_iterations, timeout_secs, + plan_item_id: args["plan_item_id"].as_str().map(str::to_string), + session_id: None, }) } @@ -156,7 +166,14 @@ impl DelegateTool { match mode { ExecutionMode::Inline => { - let config = self.parse_config_from_args(args)?; + let mut config = self.parse_config_from_args(args)?; + if config.plan_item_id.is_some() { + config.session_id = Some( + crate::agent::sub_agent::get_delegate_context() + .map_err(anyhow::Error::msg)? + .session_id, + ); + } let result = self .sub_agent_manager .run_inline(config) @@ -187,10 +204,11 @@ impl DelegateTool { } } ExecutionMode::Background => { - let config = self.parse_config_from_args(args)?; + let mut config = self.parse_config_from_args(args)?; let ctx = crate::agent::sub_agent::get_delegate_context().map_err(|_| { anyhow::anyhow!("delegate context not available: not in an agent worker") })?; + config.session_id = Some(ctx.session_id.clone()); let task_id = self .sub_agent_manager @@ -209,6 +227,7 @@ impl DelegateTool { .as_array() .ok_or_else(|| anyhow::anyhow!("parallel mode requires 'tasks' array"))?; + let ctx = crate::agent::sub_agent::get_delegate_context().ok(); let mut configs = Vec::new(); for task in tasks { let prompt = task["prompt"] @@ -228,6 +247,8 @@ impl DelegateTool { allowed_tools, max_iterations: args["max_iterations"].as_u64().map(|v| v as usize), timeout_secs: args["timeout_secs"].as_u64(), + plan_item_id: task["plan_item_id"].as_str().map(str::to_string), + session_id: ctx.as_ref().map(|ctx| ctx.session_id.clone()), }); } diff --git a/src/tools/mod.rs b/src/tools/mod.rs index 4b6379b..e9a2f9d 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -19,6 +19,7 @@ pub mod pty; pub mod registry; pub mod schema; pub mod send_message; +pub mod todo; pub mod traits; pub mod web_fetch; @@ -39,6 +40,7 @@ pub use memory::{MemoryForgetTool, MemoryRecallTool, MemoryStoreTool, TimelineRe pub use pty::{PtyManager, PtyTool}; pub use registry::ToolRegistry; pub use send_message::SendMessageTool; +pub use todo::TodoTool; pub use traits::{OutboundMessenger, Tool, ToolResult}; pub use web_fetch::WebFetchTool; @@ -54,6 +56,7 @@ use std::sync::Arc; pub fn create_default_tools( skills_loader: Arc, memory: Arc, + work_manager: Arc, sub_agent_manager: Option>, browser_config: Option<&BrowserConfig>, ) -> ToolRegistry { @@ -79,6 +82,7 @@ pub fn create_default_tools( registry.register(MemoryRecallTool::new(memory.clone())); registry.register(TimelineRecallTool::new(memory.clone())); registry.register(MemoryForgetTool::new(memory.clone())); + registry.register(TodoTool::new(work_manager)); if let Some(cfg) = browser_config && cfg.enabled diff --git a/src/tools/todo.rs b/src/tools/todo.rs new file mode 100644 index 0000000..29d8e14 --- /dev/null +++ b/src/tools/todo.rs @@ -0,0 +1,163 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::json; + +use crate::tools::traits::{Tool, ToolResult}; +use crate::work::{TaskPlan, WorkManager}; + +pub struct TodoTool { + work_manager: Arc, +} + +impl TodoTool { + pub fn new(work_manager: Arc) -> Self { + Self { work_manager } + } + + fn context() -> anyhow::Result { + crate::agent::sub_agent::get_delegate_context() + .map_err(|_| anyhow::anyhow!("todo context not available outside a session worker")) + } +} + +#[async_trait] +impl Tool for TodoTool { + fn name(&self) -> &str { + "todo" + } + + fn description(&self) -> &str { + "管理当前 session 的复杂任务计划。仅用于明确需要多步骤、跨轮次或并行委托的任务；闲聊、问答和单步操作不要创建计划。" + } + + fn parameters_schema(&self) -> serde_json::Value { + json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["create", "view", "append", "update", "close"] + }, + "objective": { "type": "string", "description": "create 时的总体目标" }, + "items": { + "type": "array", + "items": { "type": "string" }, + "description": "create 时的有序子项" + }, + "item_id": { "type": "string", "description": "update 时的 T1/T2..." }, + "title": { "type": "string", "description": "append 时的新子项标题" }, + "status": { + "type": "string", + "enum": ["pending", "in_progress", "completed", "blocked", "cancelled"], + "description": "update 的子项状态或 close 的计划状态" + }, + "summary": { "type": "string", "description": "完成结果或阻塞原因" }, + "expected_version": { "type": "integer", "description": "可选的计划版本并发校验" } + }, + "required": ["action"] + }) + } + + fn read_only(&self) -> bool { + false + } + + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + let action = args["action"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing action"))?; + let ctx = Self::context()?; + let expected = args["expected_version"].as_i64(); + let result = match action { + "create" => { + let objective = required_str(&args, "objective")?; + let items = args["items"] + .as_array() + .ok_or_else(|| anyhow::anyhow!("create requires items"))? + .iter() + .filter_map(|value| value.as_str().map(str::to_string)) + .collect::>(); + self.work_manager + .create_plan(&ctx.session_id, objective, &items) + .await + } + "view" => match self.work_manager.active_plan(&ctx.session_id).await? { + Some(plan) => Ok(plan), + None => { + return Ok(ToolResult { + success: true, + output: "当前 session 没有 active plan".to_string(), + error: None, + }); + } + }, + "append" => { + self.work_manager + .append_item(&ctx.session_id, required_str(&args, "title")?, expected) + .await + } + "update" => { + self.work_manager + .update_item( + &ctx.session_id, + required_str(&args, "item_id")?, + required_str(&args, "status")?, + args["summary"].as_str(), + expected, + ) + .await + } + "close" => { + self.work_manager + .close_plan(&ctx.session_id, required_str(&args, "status")?, expected) + .await + } + _ => return Err(anyhow::anyhow!("unknown todo action: {action}")), + }; + + match result { + Ok(plan) => Ok(ToolResult { + success: true, + output: render_plan(&plan), + error: None, + }), + Err(error) => Ok(ToolResult { + success: false, + output: String::new(), + error: Some(error.to_string()), + }), + } + } +} + +fn required_str<'a>(args: &'a serde_json::Value, key: &str) -> anyhow::Result<&'a str> { + args[key] + .as_str() + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| anyhow::anyhow!("missing required parameter: {key}")) +} + +fn render_plan(plan: &TaskPlan) -> String { + let mut output = format!( + "计划 {}（{}，version {}）\n目标：{}\n", + plan.id, plan.status, plan.version, plan.objective + ); + for item in &plan.items { + let icon = match item.status.as_str() { + "completed" => "✓", + "in_progress" => "●", + "blocked" => "!", + _ => "○", + }; + output.push_str(&format!( + "{icon} {} [{}] {}", + item.id, item.status, item.title + )); + if let Some(summary) = item.result_summary.as_deref().or(item.error.as_deref()) { + output.push_str(&format!(" — {summary}")); + } + output.push('\n'); + } + output.trim_end().to_string() +} diff --git a/src/work/mod.rs b/src/work/mod.rs new file mode 100644 index 0000000..55ea1bf --- /dev/null +++ b/src/work/mod.rs @@ -0,0 +1,689 @@ +use std::sync::Arc; + +use dashmap::DashMap; +use serde::{Deserialize, Serialize}; +use sqlx::Row; +use tokio::sync::broadcast; + +use crate::storage::{Storage, StorageError}; + +const MAX_OBJECTIVE_CHARS: usize = 2_000; +const MAX_ITEM_TITLE_CHARS: usize = 500; +const MAX_INITIAL_ITEMS: usize = 50; +const MAX_ITEMS: usize = 100; +const MAX_SUMMARY_CHARS: usize = 4_000; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct TaskItem { + pub id: String, + pub ordinal: i64, + pub title: String, + pub status: String, + pub executor_kind: Option, + pub execution_id: Option, + pub result_summary: Option, + pub error: Option, + pub version: i64, + pub updated_at: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct TaskPlan { + pub id: String, + pub session_id: String, + pub objective: String, + pub status: String, + pub version: i64, + pub created_at: i64, + pub updated_at: i64, + pub closed_at: Option, + pub items: Vec, +} + +impl TaskPlan { + pub fn compact_context(&self) -> String { + let completed = self + .items + .iter() + .filter(|item| item.status == "completed") + .count(); + let mut lines = vec![ + "## 当前任务计划".to_string(), + format!("- 目标：{}", self.objective), + format!("- 进度：{completed}/{}", self.items.len()), + ]; + let running: Vec<_> = self + .items + .iter() + .filter(|item| item.status == "in_progress") + .take(8) + .map(|item| { + let executor = item.executor_kind.as_deref().unwrap_or("main_agent"); + format!(" - {} {}（{}）", item.id, item.title, executor) + }) + .collect(); + if !running.is_empty() { + lines.push("- 正在执行：".to_string()); + lines.extend(running); + } + let blocked: Vec<_> = self + .items + .iter() + .filter(|item| item.status == "blocked") + .take(8) + .map(|item| { + format!( + " - {} {}：{}", + item.id, + item.title, + item.error.as_deref().unwrap_or("原因未知") + ) + }) + .collect(); + if !blocked.is_empty() { + lines.push("- 阻塞：".to_string()); + lines.extend(blocked); + } + if let Some(next) = self.items.iter().find(|item| item.status == "pending") { + lines.push(format!("- 下一待处理：{} {}", next.id, next.title)); + } + lines.push(format!("- 计划版本：{}", self.version)); + lines.join("\n") + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PlanChanged { + pub session_id: String, + pub reason: String, + pub changed_item_ids: Vec, + pub plan: Option, +} + +#[derive(Clone)] +pub struct WorkManager { + storage: Arc, + events: broadcast::Sender, + active_cache: Arc>>, +} + +impl WorkManager { + pub fn new(storage: Arc) -> Self { + let (events, _) = broadcast::channel(128); + Self { + storage, + events, + active_cache: Arc::new(DashMap::new()), + } + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.events.subscribe() + } + + pub async fn active_plan(&self, session_id: &str) -> Result, StorageError> { + if let Some(cached) = self.active_cache.get(session_id) { + return Ok(cached.clone()); + } + let row = sqlx::query( + "SELECT id, session_id, objective, status, version, created_at, updated_at, closed_at \ + FROM task_plans WHERE session_id = ? AND status = 'active' LIMIT 1", + ) + .bind(session_id) + .fetch_optional(self.storage.pool()) + .await?; + let Some(row) = row else { + self.active_cache.insert(session_id.to_string(), None); + return Ok(None); + }; + let plan = self.plan_from_row(row).await?; + self.cache_active_plan(&plan); + Ok(Some(plan)) + } + + pub async fn plan_for_session( + &self, + session_id: &str, + ) -> Result, StorageError> { + self.active_plan(session_id).await + } + + async fn plan_from_row(&self, row: sqlx::sqlite::SqliteRow) -> Result { + let plan_id: String = row.get("id"); + let item_rows = sqlx::query( + "SELECT id, ordinal, title, status, executor_kind, execution_id, result_summary, \ + error, version, updated_at FROM task_items WHERE plan_id = ? ORDER BY ordinal", + ) + .bind(&plan_id) + .fetch_all(self.storage.pool()) + .await?; + Ok(TaskPlan { + id: plan_id, + session_id: row.get("session_id"), + objective: row.get("objective"), + status: row.get("status"), + version: row.get("version"), + created_at: row.get("created_at"), + updated_at: row.get("updated_at"), + closed_at: row.get("closed_at"), + items: item_rows.into_iter().map(item_from_row).collect(), + }) + } + + async fn plan_by_id(&self, plan_id: &str) -> Result { + let row = sqlx::query( + "SELECT id, session_id, objective, status, version, created_at, updated_at, closed_at \ + FROM task_plans WHERE id = ?", + ) + .bind(plan_id) + .fetch_optional(self.storage.pool()) + .await? + .ok_or_else(|| StorageError::NotFound(plan_id.to_string()))?; + self.plan_from_row(row).await + } + + pub async fn create_plan( + &self, + session_id: &str, + objective: &str, + items: &[String], + ) -> Result { + if objective.trim().is_empty() || items.is_empty() { + return Err(StorageError::Conflict( + "计划目标不能为空，且至少需要一个子项".to_string(), + )); + } + validate_length("计划目标", objective, MAX_OBJECTIVE_CHARS)?; + if items.len() > MAX_INITIAL_ITEMS { + return Err(StorageError::Conflict(format!( + "初始计划最多包含 {MAX_INITIAL_ITEMS} 个子项" + ))); + } + if self.active_plan(session_id).await?.is_some() { + return Err(StorageError::Conflict( + "当前 session 已有 active plan".to_string(), + )); + } + let now = chrono::Utc::now().timestamp_millis(); + let plan_id = format!("plan-{}", crate::util::short_id()); + let mut tx = self.storage.pool().begin().await?; + sqlx::query( + "INSERT INTO task_plans (id, session_id, objective, status, version, created_at, updated_at) \ + VALUES (?, ?, ?, 'active', 1, ?, ?)", + ) + .bind(&plan_id) + .bind(session_id) + .bind(objective.trim()) + .bind(now) + .bind(now) + .execute(&mut *tx) + .await?; + for (index, title) in items.iter().enumerate() { + let title = title.trim(); + if title.is_empty() { + return Err(StorageError::Conflict("计划子项不能为空".to_string())); + } + validate_length("计划子项", title, MAX_ITEM_TITLE_CHARS)?; + sqlx::query( + "INSERT INTO task_items (id, plan_id, ordinal, title, status, version, created_at, updated_at) \ + VALUES (?, ?, ?, ?, 'pending', 1, ?, ?)", + ) + .bind(format!("T{}", index + 1)) + .bind(&plan_id) + .bind(index as i64 + 1) + .bind(title) + .bind(now) + .bind(now) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + let plan = self.plan_by_id(&plan_id).await?; + self.cache_active_plan(&plan); + self.emit("plan_created", vec![], Some(plan.clone())); + Ok(plan) + } + + pub async fn append_item( + &self, + session_id: &str, + title: &str, + expected_version: Option, + ) -> Result { + let plan = self.require_active(session_id, expected_version).await?; + if title.trim().is_empty() { + return Err(StorageError::Conflict("计划子项不能为空".to_string())); + } + if plan.items.len() >= MAX_ITEMS { + return Err(StorageError::Conflict(format!( + "计划最多包含 {MAX_ITEMS} 个子项" + ))); + } + validate_length("计划子项", title, MAX_ITEM_TITLE_CHARS)?; + let now = chrono::Utc::now().timestamp_millis(); + let ordinal = plan.items.last().map_or(1, |item| item.ordinal + 1); + let item_id = format!("T{ordinal}"); + let mut tx = self.storage.pool().begin().await?; + sqlx::query( + "INSERT INTO task_items (id, plan_id, ordinal, title, status, version, created_at, updated_at) \ + VALUES (?, ?, ?, ?, 'pending', 1, ?, ?)", + ) + .bind(&item_id) + .bind(&plan.id) + .bind(ordinal) + .bind(title.trim()) + .bind(now) + .bind(now) + .execute(&mut *tx) + .await?; + bump_plan(&mut tx, &plan.id, now).await?; + tx.commit().await?; + let updated = self.plan_by_id(&plan.id).await?; + self.cache_active_plan(&updated); + self.emit("item_added", vec![item_id], Some(updated.clone())); + Ok(updated) + } + + pub async fn update_item( + &self, + session_id: &str, + item_id: &str, + status: &str, + summary: Option<&str>, + expected_version: Option, + ) -> Result { + validate_item_status(status)?; + let plan = self.require_active(session_id, expected_version).await?; + let current = plan + .items + .iter() + .find(|item| item.id == item_id) + .ok_or_else(|| StorageError::NotFound(item_id.to_string()))?; + validate_transition(¤t.status, status)?; + let now = chrono::Utc::now().timestamp_millis(); + let summary = summary.map(|value| truncate_chars(value, MAX_SUMMARY_CHARS)); + let error = (status == "blocked") + .then_some(summary.as_deref()) + .flatten(); + let result_summary = (status == "completed") + .then_some(summary.as_deref()) + .flatten(); + let executor = match status { + "in_progress" => Some("main_agent"), + "completed" | "blocked" => current.executor_kind.as_deref(), + _ => None, + }; + let execution_id = matches!(status, "completed" | "blocked") + .then_some(current.execution_id.as_deref()) + .flatten(); + let mut tx = self.storage.pool().begin().await?; + let update = sqlx::query( + "UPDATE task_items SET status = ?, executor_kind = ?, execution_id = ?, \ + result_summary = ?, error = ?, \ + version = version + 1, updated_at = ? \ + WHERE plan_id = ? AND id = ? AND status = ? AND version = ?", + ) + .bind(status) + .bind(executor) + .bind(execution_id) + .bind(result_summary) + .bind(error) + .bind(now) + .bind(&plan.id) + .bind(item_id) + .bind(¤t.status) + .bind(current.version) + .execute(&mut *tx) + .await?; + if update.rows_affected() != 1 { + return Err(StorageError::Conflict(format!( + "子项 {item_id} 已被其他执行更新，请读取最新计划后重试" + ))); + } + bump_plan(&mut tx, &plan.id, now).await?; + tx.commit().await?; + let updated = self.plan_by_id(&plan.id).await?; + self.cache_active_plan(&updated); + self.emit( + "item_updated", + vec![item_id.to_string()], + Some(updated.clone()), + ); + Ok(updated) + } + + pub async fn assign_sub_agent( + &self, + session_id: &str, + item_id: &str, + execution_id: &str, + ) -> Result { + let plan = self.require_active(session_id, None).await?; + let item = plan + .items + .iter() + .find(|item| item.id == item_id) + .ok_or_else(|| StorageError::NotFound(item_id.to_string()))?; + if item.status != "pending" { + return Err(StorageError::Conflict(format!( + "子项 {item_id} 当前状态为 {}，不能重复委托", + item.status + ))); + } + let now = chrono::Utc::now().timestamp_millis(); + let mut tx = self.storage.pool().begin().await?; + let result = sqlx::query( + "UPDATE task_items SET status = 'in_progress', executor_kind = 'sub_agent', \ + execution_id = ?, error = NULL, version = version + 1, updated_at = ? \ + WHERE plan_id = ? AND id = ? AND status = 'pending'", + ) + .bind(execution_id) + .bind(now) + .bind(&plan.id) + .bind(item_id) + .execute(&mut *tx) + .await?; + if result.rows_affected() != 1 { + return Err(StorageError::Conflict(format!( + "子项 {item_id} 已被其他执行领取" + ))); + } + bump_plan(&mut tx, &plan.id, now).await?; + tx.commit().await?; + let updated = self.plan_by_id(&plan.id).await?; + self.cache_active_plan(&updated); + self.emit( + "item_assigned", + vec![item_id.to_string()], + Some(updated.clone()), + ); + Ok(updated) + } + + pub async fn finish_sub_agent( + &self, + session_id: &str, + item_id: &str, + execution_id: &str, + completed: bool, + summary: Option<&str>, + ) -> Result, StorageError> { + let Some(plan) = self.active_plan(session_id).await? else { + return Ok(None); + }; + let item = plan.items.iter().find(|item| item.id == item_id); + if !item.is_some_and(|item| { + item.status == "in_progress" && item.execution_id.as_deref() == Some(execution_id) + }) { + return Ok(Some(plan)); + } + let status = if completed { "completed" } else { "blocked" }; + let summary = summary.map(|value| truncate_chars(value, MAX_SUMMARY_CHARS)); + let now = chrono::Utc::now().timestamp_millis(); + let mut tx = self.storage.pool().begin().await?; + let result = sqlx::query( + "UPDATE task_items SET status = ?, result_summary = ?, error = ?, \ + version = version + 1, updated_at = ? \ + WHERE plan_id = ? AND id = ? AND execution_id = ? AND status = 'in_progress'", + ) + .bind(status) + .bind(completed.then_some(summary.as_deref()).flatten()) + .bind((!completed).then_some(summary.as_deref()).flatten()) + .bind(now) + .bind(&plan.id) + .bind(item_id) + .bind(execution_id) + .execute(&mut *tx) + .await?; + if result.rows_affected() != 1 { + drop(tx); + return Ok(Some(self.plan_by_id(&plan.id).await?)); + } + bump_plan(&mut tx, &plan.id, now).await?; + tx.commit().await?; + let updated = self.plan_by_id(&plan.id).await?; + self.cache_active_plan(&updated); + self.emit( + if completed { + "item_completed" + } else { + "item_blocked" + }, + vec![item_id.to_string()], + Some(updated.clone()), + ); + Ok(Some(updated)) + } + + pub async fn close_plan( + &self, + session_id: &str, + status: &str, + expected_version: Option, + ) -> Result { + if !matches!(status, "completed" | "cancelled") { + return Err(StorageError::Conflict( + "计划只能 completed 或 cancelled".to_string(), + )); + } + let plan = self.require_active(session_id, expected_version).await?; + if status == "completed" && plan.items.iter().any(|item| item.status != "completed") { + return Err(StorageError::Conflict( + "仍有未完成子项，不能关闭计划".to_string(), + )); + } + let now = chrono::Utc::now().timestamp_millis(); + sqlx::query( + "UPDATE task_plans SET status = ?, version = version + 1, updated_at = ?, closed_at = ? \ + WHERE id = ? AND status = 'active'", + ) + .bind(status) + .bind(now) + .bind(now) + .bind(&plan.id) + .execute(self.storage.pool()) + .await?; + let closed = self.plan_by_id(&plan.id).await?; + self.active_cache.insert(session_id.to_string(), None); + self.emit("plan_closed", vec![], Some(closed.clone())); + Ok(closed) + } + + async fn require_active( + &self, + session_id: &str, + expected_version: Option, + ) -> Result { + let plan = self + .active_plan(session_id) + .await? + .ok_or_else(|| StorageError::NotFound("active plan".to_string()))?; + if expected_version.is_some_and(|version| version != plan.version) { + return Err(StorageError::Conflict(format!( + "计划版本已变化：expected {}, actual {}", + expected_version.unwrap_or_default(), + plan.version + ))); + } + Ok(plan) + } + + fn emit(&self, reason: &str, changed_item_ids: Vec, plan: Option) { + let Some(session_id) = plan.as_ref().map(|plan| plan.session_id.clone()) else { + return; + }; + let _ = self.events.send(PlanChanged { + session_id, + reason: reason.to_string(), + changed_item_ids, + plan, + }); + } + + fn cache_active_plan(&self, plan: &TaskPlan) { + let session_id = plan.session_id.clone(); + match self.active_cache.entry(session_id) { + dashmap::mapref::entry::Entry::Occupied(mut entry) => { + let should_replace = entry + .get() + .as_ref() + .is_none_or(|cached| cached.id != plan.id || cached.version <= plan.version); + if should_replace { + entry.insert(Some(plan.clone())); + } + } + dashmap::mapref::entry::Entry::Vacant(entry) => { + entry.insert(Some(plan.clone())); + } + } + } +} + +fn item_from_row(row: sqlx::sqlite::SqliteRow) -> TaskItem { + TaskItem { + id: row.get("id"), + ordinal: row.get("ordinal"), + title: row.get("title"), + status: row.get("status"), + executor_kind: row.get("executor_kind"), + execution_id: row.get("execution_id"), + result_summary: row.get("result_summary"), + error: row.get("error"), + version: row.get("version"), + updated_at: row.get("updated_at"), + } +} + +async fn bump_plan( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + plan_id: &str, + now: i64, +) -> Result<(), sqlx::Error> { + sqlx::query("UPDATE task_plans SET version = version + 1, updated_at = ? WHERE id = ?") + .bind(now) + .bind(plan_id) + .execute(&mut **tx) + .await?; + Ok(()) +} + +fn validate_item_status(status: &str) -> Result<(), StorageError> { + if matches!(status, "pending" | "in_progress" | "completed" | "blocked") { + Ok(()) + } else { + Err(StorageError::Conflict(format!("未知子项状态：{status}"))) + } +} + +fn validate_transition(from: &str, to: &str) -> Result<(), StorageError> { + let allowed = matches!( + (from, to), + ("pending", "in_progress") + | ("pending", "completed") + | ("pending", "blocked") + | ("in_progress", "completed") + | ("in_progress", "blocked") + | ("blocked", "pending") + | ("blocked", "in_progress") + | ("blocked", "completed") + ) || from == to; + if allowed { + Ok(()) + } else { + Err(StorageError::Conflict(format!( + "不允许的状态迁移：{from} -> {to}" + ))) + } +} + +fn validate_length(label: &str, value: &str, max_chars: usize) -> Result<(), StorageError> { + if value.chars().count() <= max_chars { + Ok(()) + } else { + Err(StorageError::Conflict(format!( + "{label}不能超过 {max_chars} 个字符" + ))) + } +} + +fn truncate_chars(value: &str, max_chars: usize) -> String { + value.chars().take(max_chars).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn manager() -> (WorkManager, tempfile::TempDir) { + let dir = tempfile::tempdir().unwrap(); + let storage = Arc::new(Storage::new(&dir.path().join("work.db")).await.unwrap()); + (WorkManager::new(storage), dir) + } + + #[tokio::test] + async fn one_active_plan_per_session_and_parallel_items() { + let (manager, _dir) = manager().await; + let plan = manager + .create_plan("cli:a:d1", "ship it", &["one".into(), "two".into()]) + .await + .unwrap(); + assert_eq!(plan.items.len(), 2); + let other = manager + .create_plan("cli:b:d2", "other", &["one".into()]) + .await + .unwrap(); + assert_eq!(other.items[0].id, "T1"); + assert!( + manager + .create_plan("cli:a:d1", "another", &["x".into()]) + .await + .is_err() + ); + + manager + .assign_sub_agent("cli:a:d1", "T1", "run-1") + .await + .unwrap(); + assert!( + manager + .assign_sub_agent("cli:a:d1", "T1", "run-duplicate") + .await + .is_err() + ); + let plan = manager + .assign_sub_agent("cli:a:d1", "T2", "run-2") + .await + .unwrap(); + assert_eq!( + plan.items + .iter() + .filter(|item| item.status == "in_progress") + .count(), + 2 + ); + } + + #[tokio::test] + async fn late_sub_agent_result_does_not_mutate_cancelled_plan() { + let (manager, _dir) = manager().await; + manager + .create_plan("cli:a:d1", "ship", &["one".into()]) + .await + .unwrap(); + manager + .assign_sub_agent("cli:a:d1", "T1", "run-1") + .await + .unwrap(); + manager + .close_plan("cli:a:d1", "cancelled", None) + .await + .unwrap(); + assert!( + manager + .finish_sub_agent("cli:a:d1", "T1", "run-1", true, Some("done")) + .await + .unwrap() + .is_none() + ); + } +} diff --git a/tests/test_request_format.rs b/tests/test_request_format.rs index 0963223..fc41047 100644 --- a/tests/test_request_format.rs +++ b/tests/test_request_format.rs @@ -1,5 +1,6 @@ use picobot::protocol::{HistoryMessage, SessionSummary, WsInbound, WsOutbound}; use picobot::providers::{ChatCompletionRequest, Message}; +use picobot::work::{TaskItem, TaskPlan}; /// Test that message with special characters is properly escaped #[test] @@ -147,3 +148,55 @@ fn test_bounded_session_history_protocol() { other => panic!("unexpected decoded variant: {other:?}"), } } + +#[test] +fn test_session_plan_protocol_is_structured_and_versioned() { + let inbound = WsInbound::GetSessionPlan { + session_id: "cli_chat:client:dialog".to_string(), + }; + assert!( + serde_json::to_string(&inbound) + .unwrap() + .contains(r#""type":"get_session_plan""#) + ); + + let plan = TaskPlan { + id: "plan-1".to_string(), + session_id: "cli_chat:client:dialog".to_string(), + objective: "实现 Todo".to_string(), + status: "active".to_string(), + version: 3, + created_at: 1, + updated_at: 2, + closed_at: None, + items: vec![TaskItem { + id: "T1".to_string(), + ordinal: 1, + title: "实现协议".to_string(), + status: "in_progress".to_string(), + executor_kind: Some("sub_agent".to_string()), + execution_id: Some("run-1".to_string()), + result_summary: None, + error: None, + version: 2, + updated_at: 2, + }], + }; + let outbound = WsOutbound::PlanUpdated { + session_id: plan.session_id.clone(), + reason: "item_assigned".to_string(), + changed_item_ids: vec!["T1".to_string()], + plan: Some(plan), + }; + let decoded: WsOutbound = + serde_json::from_str(&serde_json::to_string(&outbound).unwrap()).unwrap(); + match decoded { + WsOutbound::PlanUpdated { plan, reason, .. } => { + let plan = plan.unwrap(); + assert_eq!(plan.version, 3); + assert_eq!(plan.items[0].execution_id.as_deref(), Some("run-1")); + assert_eq!(reason, "item_assigned"); + } + other => panic!("unexpected decoded variant: {other:?}"), + } +} diff --git a/webui/src/pages/ChatPage.svelte b/webui/src/pages/ChatPage.svelte index 5ccaa8f..8f4446f 100644 --- a/webui/src/pages/ChatPage.svelte +++ b/webui/src/pages/ChatPage.svelte @@ -15,11 +15,16 @@ let selectedCommand = $state(0); let commandMenuDismissed = $state(false); let thinking = $state(false); + let plansBySession = $state({}); + let unseenPlanSessions = $state({}); + let todoOpen = $state(false); let messageBox; let input; let reconnectTimer; let stopped = false; const currentSession = $derived(sessions.find((item) => item.session_id === currentId)); + const currentPlan = $derived(currentId ? plansBySession[currentId] || null : null); + const completedItems = $derived(currentPlan?.items?.filter((item) => item.status === "completed").length || 0); const filteredSessions = $derived(sessions.filter((item) => item.title.toLowerCase().includes(search.toLowerCase()))); const commandQuery = $derived( !commandMenuDismissed && draft.startsWith("/") && !/[\s]/.test(draft) @@ -50,6 +55,9 @@ socket = ws; ws.onopen = () => { connected = true; + plansBySession = {}; + unseenPlanSessions = {}; + todoOpen = false; send({ type: "list_sessions", include_archived: false }); send({ type: "get_slash_commands" }); }; @@ -77,6 +85,7 @@ case "session_loaded": currentId = frame.session_id; send({ type: "get_session_history", session_id: currentId, limit: 1000 }); + send({ type: "get_session_plan", session_id: currentId }); break; case "session_history": if (frame.session_id === currentId) { @@ -84,6 +93,30 @@ scrollToBottom(); } break; + case "session_plan": + if (frame.plan) { + const previous = plansBySession[frame.session_id]; + if (frame.plan.id !== previous?.id || frame.plan.version >= (previous?.version || 0)) { + plansBySession[frame.session_id] = frame.plan; + } + } else if (plansBySession[frame.session_id]?.status !== "active") { + plansBySession[frame.session_id] = null; + } + break; + case "plan_updated": { + const previousPlan = plansBySession[frame.session_id]; + const previousVersion = previousPlan?.version || 0; + const nextVersion = frame.plan?.version || 0; + if (frame.plan?.id === previousPlan?.id && nextVersion && nextVersion <= previousVersion) break; + plansBySession[frame.session_id] = frame.plan; + if (frame.session_id === currentId) { + todoOpen = true; + unseenPlanSessions[frame.session_id] = false; + } else { + unseenPlanSessions[frame.session_id] = true; + } + break; + } case "slash_commands_list": commands = frame.commands || []; selectedCommand = 0; @@ -115,9 +148,18 @@ if (!id) return; currentId = id; messages = []; + todoOpen = Boolean(unseenPlanSessions[id]); + unseenPlanSessions[id] = false; send({ type: "load_session", session_id: id }); } + function itemIcon(status) { + if (status === "completed") return "✓"; + if (status === "in_progress") return "●"; + if (status === "blocked") return "!"; + return "○"; + } + function submit() { const content = draft.trim(); if (!content || !connected) return; @@ -200,14 +242,14 @@ }); -
+
diff --git a/webui/src/styles.css b/webui/src/styles.css index b4b5b21..a0d3181 100644 --- a/webui/src/styles.css +++ b/webui/src/styles.css @@ -42,7 +42,8 @@ main { min-width: 0; height: 100vh; display: flex; flex-direction: column; } .topbar p { font-size: 12px; color: var(--muted); margin: 4px 0 0; } .topbar .menu { display: none; background: none; border: 0; color: var(--text); font-size: 20px; margin-right: 12px; } .page { min-height: 0; flex: 1; } -.chat-layout { display: grid; grid-template-columns: 270px 1fr; } +.chat-layout { display: grid; grid-template-columns: 270px minmax(0, 1fr); } +.chat-layout.todo-open { grid-template-columns: 270px minmax(0, 1fr) 320px; } .sessions-panel { border-right: 1px solid var(--line); padding: 16px; background: rgb(14 17 21 / 66%); overflow: auto; } .primary, .secondary { border-radius: 9px; padding: 9px 14px; font-weight: 650; cursor: pointer; } .primary { border: 1px solid var(--accent); background: var(--accent); color: #11160a; } @@ -57,11 +58,15 @@ main { min-width: 0; height: 100vh; display: flex; flex-direction: column; } .session-item:hover, .session-item.active { background: var(--panel-2); border-color: var(--line); } .session-item.active { border-left-color: var(--accent); } .session-item strong { font-size: 13px; display: block; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; } +.plan-unread { display: inline-block; width: 7px; height: 7px; margin-left: 7px; border-radius: 50%; background: var(--accent); box-shadow: 0 0 9px var(--accent); } .session-item small { color: var(--muted); font-size: 11px; display: flex; justify-content: space-between; margin-top: 6px; } .chat-panel { display: flex; flex-direction: column; min-width: 0; min-height: 0; } .chat-heading { height: 57px; flex: 0 0 57px; border-bottom: 1px solid var(--line); display: flex; align-items: center; justify-content: space-between; padding: 0 20px; } .chat-heading strong, .chat-heading small { display: block; } .chat-heading strong { font-size: 13px; } +.chat-heading-actions { display: flex; align-items: center; gap: 9px; } +.todo-toggle { border: 1px solid var(--line); border-radius: 8px; padding: 6px 9px; background: var(--panel-2); color: var(--muted); font-size: 11px; cursor: pointer; } +.todo-toggle.active { border-color: #657d32; color: var(--accent); } .chat-heading small { font-size: 10px; color: var(--muted); margin-top: 3px; } .icon-button { border: 0; background: none; color: var(--muted); font-size: 20px; cursor: pointer; border-radius: 6px; } .tooltip { z-index: 50; border: 1px solid #3c4652; background: #222831; color: var(--text); padding: 6px 9px; border-radius: 7px; font-size: 11px; box-shadow: 0 8px 24px #0008; } @@ -85,6 +90,29 @@ main { min-width: 0; height: 100vh; display: flex; flex-direction: column; } .send { width: 36px; height: 36px; border-radius: 10px; background: var(--accent); border: 0; font-size: 19px; cursor: pointer; } .composer small { grid-column: 1 / -1; display: flex; justify-content: space-between; padding: 3px 7px; color: var(--muted); font-size: 10px; } .composer small .online { color: var(--accent); } +.todo-panel { display: none; min-width: 0; border-left: 1px solid var(--line); background: #0e1115; overflow: hidden; } +.todo-open .todo-panel { display: flex; flex-direction: column; } +.todo-heading { min-height: 78px; padding: 14px 14px 12px 16px; border-bottom: 1px solid var(--line); display: flex; align-items: flex-start; justify-content: space-between; gap: 10px; } +.todo-heading small, .todo-heading strong { display: block; } +.todo-heading small { color: var(--muted); font-size: 10px; margin-bottom: 5px; } +.todo-heading strong { font-size: 13px; line-height: 1.4; } +.todo-progress { height: 3px; background: #252b32; } +.todo-progress span { display: block; height: 100%; background: var(--accent); transition: width .2s; } +.todo-summary { display: flex; justify-content: space-between; padding: 11px 16px; color: var(--muted); font-size: 10px; } +.todo-items { min-height: 0; overflow-y: auto; padding: 4px 12px 18px; } +.todo-item { display: grid; grid-template-columns: 24px 1fr; gap: 8px; padding: 10px 8px; border-radius: 9px; color: #b8bec6; } +.todo-item + .todo-item { margin-top: 3px; } +.todo-item.running { background: #202519; color: var(--text); } +.todo-item.blocked { background: #28191c; } +.todo-item.done { color: #727a83; } +.todo-icon { width: 21px; height: 21px; display: grid; place-items: center; border: 1px solid var(--line); border-radius: 50%; font-size: 11px; } +.todo-item.running .todo-icon { color: var(--accent); border-color: #657d32; } +.todo-item.blocked .todo-icon { color: #ff8d8d; border-color: #74363b; } +.todo-item.done .todo-icon { color: var(--accent); } +.todo-item strong, .todo-item small { display: block; } +.todo-item strong { font-size: 12px; line-height: 1.45; } +.todo-item small { color: var(--muted); font-size: 10px; margin-top: 3px; } +.todo-item p { margin: 6px 0 0; color: #aab1ba; font-size: 10px; line-height: 1.45; white-space: pre-wrap; overflow-wrap: anywhere; } .command-menu { position: absolute; z-index: 5; left: 0; right: 0; bottom: calc(100% + 8px); max-height: min(360px, 48vh); overflow-y: auto; padding: 6px; border: 1px solid #3b454f; border-radius: 13px; background: #14181d; box-shadow: 0 18px 50px #000b; } .command-menu-heading { position: sticky; top: -6px; z-index: 1; display: flex; justify-content: space-between; gap: 16px; padding: 9px 10px 8px; color: var(--muted); background: #14181df2; font-size: 10px; } .command-menu-heading kbd { color: #89919a; font: inherit; } @@ -146,7 +174,9 @@ code { color: var(--accent); } .sidebar.open { transform: none; } .topbar .menu { display: block; } .chat-layout { grid-template-columns: 1fr; } + .chat-layout.todo-open { grid-template-columns: 1fr; } .sessions-panel { display: none; } + .todo-open .todo-panel { position: fixed; z-index: 20; top: 76px; right: 0; bottom: 0; width: min(340px, 92vw); box-shadow: -18px 0 50px #000b; } .content-page { padding: 16px; } .settings-grid { grid-template-columns: 1fr; } .settings-nav { flex-direction: row; overflow: auto; }