diff --git a/AGENTS.md b/AGENTS.md index 24f9604..94c86a9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -93,7 +93,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **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, drains same-Turn steering only at safe model boundaries, calls LLM providers, executes tools, and returns one result -- **AgentCatalog** is immutable per runtime generation; when orchestration is enabled, candidate preparation strictly validates trusted Markdown definitions, Provider profiles, delegated tools, Skill allowlists, and delegation edges before activation. Named Agents currently support foreground execution; durable named background delivery is not available until the run/inbox phase lands +- **AgentCatalog** is immutable per runtime generation; when orchestration is enabled, candidate preparation strictly validates trusted Markdown definitions, Provider profiles, delegated tools, Skill allowlists, and delegation edges before activation. Named Agents support foreground (single/batch) and Root-initiated single background execution with durable run/inbox delivery; background batches, nested background, and legacy general background (kept for one transition version) remain restricted - **WebUI management APIs** only expose allowlisted config/profile/log/storage operations; keep response limits, secret redaction, atomic config writes, and profile path allowlists intact - **WebUI styling** uses the local Fluent 2 semantic tokens in `webui/src/styles.css`; page and component styles should consume the aliases instead of introducing independent hard-coded palettes, and must preserve selectable light/dark and brand-color themes, keyboard focus, responsive layout, and reduced-motion behavior. Browser-only appearance settings belong in `lib/theme.js`/`localStorage`, and `public/theme-init.js` must restore them before Svelte mounts - **Gateway config reload** validates and prepares a complete next runtime generation before retiring the old one; close runtime admission before draining inbound lanes, Sessions, Scheduler jobs, and background sub-agents; MCP connects only during activation; host/port, workspace, and effective storage path remain restart-only, and runtime reload must never mutate process environment variables diff --git a/README.md b/README.md index 9bad75b..655b12e 100644 --- a/README.md +++ b/README.md @@ -344,6 +344,8 @@ PicoBot 有两类记忆： | `memory_store` / `memory_recall` / `timeline_recall` / `memory_forget` | 长期记忆操作 | | `reload_config` | 在用户明确要求时校验并重新加载 Gateway 配置 | | `delegate` | 向具名 Agent 委托单个或批量任务；`foreground` 等待结果，`background` 异步执行。批量 foreground 会并发运行并按请求顺序聚合 | +| `agent_task` | 查询/列出/读取结果/取消已持久化的具名 Agent run（仅编排启用时注册） | +| `emit_signal` | 后台 run 向主 Agent 发送结构化内部信号（queue/steer 投递；仅带 signal 契约的 run 注册） | | `todo` | 为复杂、多轮任务创建并更新当前 session 的持久化计划 | | `send_message` | 向指定渠道或当前会话发送消息，可附带文件/截图；WebUI/TUI 当前 Turn 的附件并入最终回复 | | `chat_manager` | 查看渠道、会话和历史消息 | @@ -406,7 +408,7 @@ Skill 是包含 `SKILL.md` 的目录。加载优先级从高到低： ### 具名子 Agent（Phase 1） -启用 `agent_orchestration.enabled` 后，PicoBot 会在 Gateway 候选运行代构造时从配置目录下的 `definitions_dir` 加载 `*.md`。相对路径按 `config.json` 所在目录解析，且 canonical path 不得逃逸该目录；角色文件、Provider profile、工具、Skill 和委托边任一无效都会拒绝启动或热重载。第一阶段只开放具名 `foreground` 执行；具名 `background` 和 background 批量接纳会明确拒绝，直到 durable run/inbox 阶段完成。未启用时，旧 general 单任务 background 仍作为兼容路径存在。 +启用 `agent_orchestration.enabled` 后，PicoBot 会在 Gateway 候选运行代构造时从配置目录下的 `definitions_dir` 加载 `*.md`。相对路径按 `config.json` 所在目录解析，且 canonical path 不得逃逸该目录；角色文件、Provider profile、工具、Skill 和委托边任一无效都会拒绝启动或热重载。支持具名 `foreground`（单/批量）与 Root 发起的具名 `background` 单任务：background 结果经 durable inbox 由主 Agent 的 continuation Turn 汇总，可配合 `emit_signal`（queue/steer）推送内部信号。background 批量与子 Agent 发起的 background 尚未开放；未启用时，旧 general 单任务 background 仍作为兼容路径存在（带迁移提示）。 ```md --- diff --git a/build.rs b/build.rs index 2687d71..be45325 100644 --- a/build.rs +++ b/build.rs @@ -13,6 +13,29 @@ fn main() { let skills_out_dir = Path::new(&out_dir).join("skills"); fs::create_dir_all(&skills_out_dir).unwrap(); + println!("cargo:rerun-if-changed=resources/agents"); + let agents_dir = Path::new("resources/agents"); + let agents_out_dir = Path::new(&out_dir).join("agents"); + fs::create_dir_all(&agents_out_dir).unwrap(); + let mut agents = Vec::new(); + if let Ok(entries) = fs::read_dir(agents_dir) { + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("md") { + continue; + } + let agent_name = path + .file_stem() + .unwrap() + .to_str() + .unwrap() + .to_string(); + fs::copy(&path, agents_out_dir.join(format!("{agent_name}.md"))).unwrap(); + agents.push(agent_name); + } + } + agents.sort(); + let mut skills = Vec::new(); if let Ok(entries) = fs::read_dir(skills_dir) { @@ -58,6 +81,29 @@ pub static EMBEDDED_SKILLS: &[EmbeddedSkill] = &[ let generated_path = Path::new(&out_dir).join("embedded_skills.rs"); let mut f = fs::File::create(&generated_path).unwrap(); f.write_all(code.as_bytes()).unwrap(); + + let mut agent_code = String::from( + r#"pub struct EmbeddedAgent { + pub name: &'static str, + pub content: &'static str, +} + +pub static EMBEDDED_AGENTS: &[EmbeddedAgent] = &[ +"#, + ); + for name in &agents { + let file_path = agents_out_dir + .join(format!("{name}.md")) + .to_string_lossy() + .to_string(); + agent_code.push_str(&format!( + " EmbeddedAgent {{ name: \"{name}\", content: include_str!(\"{file_path}\") }},\n", + )); + } + agent_code.push_str("];\n"); + let agent_path = Path::new(&out_dir).join("embedded_agents.rs"); + let mut f = fs::File::create(&agent_path).unwrap(); + f.write_all(agent_code.as_bytes()).unwrap(); } fn build_webui(out_dir: &Path) { diff --git a/resources/agents/general-purpose.md b/resources/agents/general-purpose.md new file mode 100644 index 0000000..45d2938 --- /dev/null +++ b/resources/agents/general-purpose.md @@ -0,0 +1,34 @@ +--- +id: general-purpose +description: 通用目的子代理，处理主 Agent 委托的独立子任务 +llm_profile: default +tools: + - bash + - file_read + - file_search + - content_search + - web_fetch + - calculator + - sleep +--- + +# General Purpose Agent + +You are the **general-purpose** sub-agent of PicoBot, a general-purpose assistant. + +## Role +- Handle any independent subtask delegated by the main agent. +- Work autonomously and report back concrete results. + +## Principles +- Follow the task description and context provided by the delegator. +- Use the tools you need; prefer read-only operations unless the task requires changes. +- Be accurate: do not fabricate tool results or guesses; report failures honestly. +- If information is missing, ask or state the gap rather than inventing it. +- You do not hold the main session's conversation history; rely on the context given in your task. +- Do not delegate further to other agents unless explicitly instructed. + +## Output +- Deliver the final result in a clear, structured, self-contained format. +- Keep it concise unless the task requires detail. +- If the task cannot be completed, explain why and what is blocking it. diff --git a/resources/skills/about-picobot/references/architecture.md b/resources/skills/about-picobot/references/architecture.md index 395517e..6377cf5 100644 --- a/resources/skills/about-picobot/references/architecture.md +++ b/resources/skills/about-picobot/references/architecture.md @@ -50,7 +50,7 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message - 每个活动 Turn 独占一个 TurnSink；平台 message ID 和 reaction 清理状态只存在于 sink 内 - Tools 接收原始参数，通常返回字符串结果；有状态适配器额外接收 session/turn `ToolExecutionContext` - MCP 工具在 Gateway 初始化时连接服务器、发现工具，并包装成普通 Tool 注册到 ToolRegistry -- 具名子 Agent 从运行代不可变 `AgentCatalog` 加载，Definition 固定 Provider profile、工具/Skill allowlist、委托边与限制；新工具默认 RootOnly。Phase 1 支持单个/批量 foreground（批量并发、按请求顺序返回）和显式父子授权；具名 background 在 durable run/inbox 完成前拒绝。禁用编排时旧 general background 仍通过 MessageBus 直接通知原会话 +- 具名子 Agent 从运行代不可变 `AgentCatalog` 加载，Definition 固定 Provider profile、工具/Skill allowlist、委托边与限制；新工具默认 RootOnly。支持单个/批量 foreground（批量并发、按请求顺序返回）和显式父子授权；Root 对具名 Agent 的 background 单任务走 durable run/inbox + continuation 投递（结果不再直接通知 Channel）。禁用编排时旧 general background 仍通过 MessageBus 直接通知原会话 - 复杂任务可使用 `todo` 创建 session 级计划；多个子项可通过 `delegate.plan_item_id` 并行委托，子 Agent 不能修改计划 - WebUI 聊天复用 `/ws` 与 `cli_chat`；同源管理 API 只读取受限的日志、任务、记忆，并对白名单配置文件做原子写入 - WebUI 使用 Svelte 5 + Vite，Bits UI 提供无样式可访问组件；`cargo build` 增量生成前端到 Cargo `OUT_DIR`，再嵌入单二进制，仓库不保存生成产物 @@ -258,14 +258,66 @@ Gateway 初始化时读取 `config.mcp.servers`： | 模式 | 行为 | |------|------| -| `foreground` | 当前轮等待一个或多个子 Agent；批量任务并发执行并按请求顺序聚合 | -| `background` | 异步执行并返回 run ID；当前只有旧 general 兼容路径可用 | +| `foreground` | 当前轮等待一个或多个子 Agent；批量任务并发执行并按请求顺序聚合，全部持久化到 `agent_runs` | +| `background` | 异步执行并立即返回 run ID；仅限 Root 对具名 Agent 的单任务，结果经 durable inbox 由主 Agent 的 continuation Turn 汇总 | -启用 `agent_orchestration` 后，具名 Definition 固定角色、Provider profile、工具/Skill allowlist、委托边和限制。`allowed_tools` 只能收窄 Definition，不能扩权；新工具默认 RootOnly，当前明确可委托的工具包括 `file_read`、`file_search`、`content_search`、`web_fetch`、`calculator`、普通 `browser` 动作和 `sleep`。具名 Agent可按委托图继续 foreground 委托，但 ancestry 重复、越深度或不在白名单的目标会拒绝。具名 background 需要后续 durable run/inbox，当前明确拒绝。 +### 具名 Agent Definition（身份设定） -未启用编排或省略 target 时使用旧 general 兼容路径。其工具也只能取旧默认集合与 Delegatable 策略的交集；旧后台任务写入 `background_tasks` 表，完成后通过原 channel/chat 直接通知，默认 24 小时后清理，不具备 durable inbox 语义。 +启用 `agent_orchestration` 后，每个具名子 Agent 是一个 Markdown 文件：`<配置目录>/agents/.md`（默认 `~/.picobot/agents/`）。frontmatter 只保存非秘密引用与限制（API key/base URL 仍在 `config.json`/`.env`）： -后台子 Agent 通过 `TaskSupervisor::spawn_graceful` 注册，受 `gateway.max_concurrent_background_tasks` 限制；Gateway 关停时先收到取消信号，再在总宽限期内清理。 +```md +--- +id: researcher +description: 搜索、阅读并整理技术资料 +llm_profile: research-sonnet + +tools: + - file_read + - file_search + - content_search + - web_fetch + +delegates: + - reviewer + +skills: + - technical-research + +limits: + timeout_secs: 900 + max_iterations: 24 + max_children: 4 + max_depth: 3 + +signal: + delivery: steer + severity_allowlist: [info, warning, critical] +--- + +# Role + +你是一名严谨的研究 Agent。只返回与任务有关的结论、证据和不确定性。 +``` + +- `id`：`[a-z][a-z0-9_-]{0,63}`，文件名必须与 id 一致；`root`/`main`/`default`/`general` 为保留名。重复 ID、大小写折叠冲突、越界 symlink 或引用错误（未知 Provider profile、未注册/不可委托工具、未知 skill 或 delegate 目标）会拒绝整个候选运行代，绝不静默裁剪。 +- `llm_profile`：引用 `config.json` 中 `agents` key，Definition 绑定 Provider 与模型，运行中不热切换。 +- `tools`/`skills`：固定 allowlist。`skills` 声明要求工具集含 `get_skill`，且只注入该 allowlist。新工具默认 RootOnly；当前可委托工具为 `file_read`、`file_search`、`content_search`、`web_fetch`、`calculator`、普通 `browser` 动作和 `sleep`。 +- `delegates`：出边白名单，运行时还校验 ancestry 重复、`max_tree_depth` 与树级 `max_runs_per_tree` 预算。 +- `signal`：可选信号契约。带该块的 run 才获得 `emit_signal` 工具（fail-closed）；`delivery: steer` 使信号在活动 Turn 的安全边界注入主 Agent，`queue` 走 continuation。 +- 角色正文（`---` 之后）即 `role_prompt`，与 frontmatter 一起做 SHA-256 `definition_hash` 快照。 + +### 执行与投递 + +- 每次具名委托先持久化 run（含 execution_id、budget、Definition 快照），`allowed_tools` 只能收窄、不能扩权；子 Agent 输出视为不可信数据。 +- foreground 父 run 等待子 run 时进入 `waiting_children` 且不占 provider/tool step permit；run quota 只约束 background 接纳，嵌套 foreground 并发上限为 1 时不死锁。 +- background 接纳时预留 completion slot（容量条件更新），runner 持有 run quota permit 与 activity guard 直到 terminal commit；完成后 completion 事件落 `agent_inbox_events`，Session worker 按 `max_user_turn_burst_before_inbox`/`max_inbox_wait_secs` 公平调度，以 hidden trigger + 只读工具集的 continuation Turn 让主 Agent 汇总结果，不再直接发 Channel 通知。失败按 lease token 释放重试，超 `max_inbox_delivery_attempts` 进 dead-letter，重启经 activation recovery 收敛。 +- `agent_task`（get/list/get_result/cancel）查询与控制 run；`agent_task.cancel` 会把该 run 未消费的普通信号标记 superseded。`/stop` 取消活动 run 但保留已存在的 pending 事件；archive/delete 取消 run 并将未消费事件 dead-letter。 +- 后台 run 内可调用 `emit_signal`（key/severity/summary/details/dedupe_key），总数、速率、burst、severity allowlist、载荷大小/深度与冷却窗去重均由契约强制；steer 信号经两阶段 admission（claim → mailbox 预留 → admit(turn_id) → 激活）注入当前 Turn，`/stop` 时按 token 条件释放回 pending，绝不静默丢弃。 +- 每个 run 的完成事件 payload 携带该 run 已发出的 signal IDs，主 Agent 可识别重复报告。 + +未启用编排或省略 target 时使用旧 general 兼容路径（结果带迁移提示）。其工具也只能取旧默认集合与 Delegatable 策略的交集；旧后台任务写入 `background_tasks` 表，完成后通过原 channel/chat 直接通知，默认 24 小时后清理，不具备 durable inbox 语义，等待一个版本观察后随旧适配器移除。 + +后台子 Agent 通过 `TaskSupervisor::spawn_graceful` 注册；Gateway 关停时先收到取消信号，再在总宽限期内清理。 ## Session Todo 计划 diff --git a/resources/skills/about-picobot/references/config.md b/resources/skills/about-picobot/references/config.md index 9c60add..7964ea7 100644 --- a/resources/skills/about-picobot/references/config.md +++ b/resources/skills/about-picobot/references/config.md @@ -62,17 +62,17 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `definitions_dir` | agents | 第一层 `*.md` Definition 目录 | | `root_delegates` | [] | Root 可委托的具名 Agent ID | | `max_tree_depth` | 4 | Root-relative 委托深度硬上限 | -| `max_runs_per_tree` | 16 | 单任务树 run 预算（Phase 2 持久 Coordinator 强制完整树计数） | -| `max_concurrent_runs` / `max_concurrent_runs_per_session` | 6 / 4 | run admission 上限（Phase 2） | -| `max_concurrent_provider_steps` / `..._per_session` | 8 / 4 | Provider step 上限（Phase 2） | -| `max_concurrent_tool_steps` / `..._per_session` | 16 / 8 | 普通工具 step 上限（Phase 2） | -| `max_pending_inbox_events_per_session` | 128 | durable inbox 容量（Phase 3） | -| `inbox_event_ttl_hours` | 168 | inbox event TTL（Phase 3） | -| `max_inbox_delivery_attempts` | 8 | inbox 最大投递次数（Phase 3） | -| `max_user_turn_burst_before_inbox` | 4 | 用户 Turn 公平调度阈值（Phase 3） | -| `max_inbox_wait_secs` | 30 | inbox 最大等待阈值（Phase 3） | +| `max_runs_per_tree` | 16 | 单任务树 run 预算（树级原子计数强制） | +| `max_concurrent_runs` / `max_concurrent_runs_per_session` | 6 / 4 | background run 接纳配额（global→session 顺序获取，runner 持有至 terminal commit；foreground 不占） | +| `max_concurrent_provider_steps` / `..._per_session` | 8 / 4 | Provider step 上限（global→session） | +| `max_concurrent_tool_steps` / `..._per_session` | 16 / 8 | 普通工具 step 上限（global→session） | +| `max_pending_inbox_events_per_session` | 128 | durable inbox 容量（条件更新，预留槽不可被信号挤占） | +| `inbox_event_ttl_hours` | 168 | inbox 事件 TTL（已预留配置；TTL 清理尚未实现） | +| `max_inbox_delivery_attempts` | 8 | inbox 最大投递次数，超限 dead-letter | +| `max_user_turn_burst_before_inbox` | 4 | 用户 Turn 公平调度阈值 | +| `max_inbox_wait_secs` | 30 | inbox 最大等待阈值 | -当前已实现具名 foreground Agent、不同 `llm_profile`、固定工具/Skill allowlist、批量并发和父子委托边校验。具名 background 会明确拒绝，直到 durable run/inbox 实现；未启用时旧 general background 兼容路径保持可用。 +已实现：具名 foreground 与 background（Root 单任务）、不同 `llm_profile`、固定工具/Skill allowlist、批量并发、父子委托边校验、durable inbox continuation、`emit_signal`（queue/steer）、run quota 与 step gate。未开放：background 批量、子 Agent 发起的 background、`idempotency_key` 工具入口。未启用编排时旧 general background 兼容路径保持可用（带迁移提示，等待一个版本观察后移除）。 ## gateway 字段 diff --git a/resources/skills/about-picobot/references/db-schema.md b/resources/skills/about-picobot/references/db-schema.md index 4229e29..918ce49 100644 --- a/resources/skills/about-picobot/references/db-schema.md +++ b/resources/skills/about-picobot/references/db-schema.md @@ -2,7 +2,7 @@ 数据库为 SQLite，默认位于 workspace 下的 `picobot.db`。 -连接启用 WAL、`synchronous=NORMAL`、foreign keys、5 秒 busy timeout，连接池最多 8 个连接。当前 `PRAGMA user_version=4`；启动时会在事务内补齐旧库字段和索引，遇到比程序更新的 schema version 会拒绝启动。 +连接启用 WAL、`synchronous=NORMAL`、foreign keys、5 秒 busy timeout，连接池最多 8 个连接。当前 `PRAGMA user_version=6`；启动时会在事务内补齐旧库字段和索引，遇到比程序更新的 schema version 会拒绝启动。 ## sessions 表 @@ -23,6 +23,8 @@ | `deleted_at` | INTEGER | 软删除时间戳 | | `last_consolidated_at` | INTEGER | 上次记忆归并时间 | | `last_compressed_message_at` | INTEGER | 上次上下文压缩边界时间戳 | +| `delivery_context` | TEXT | 渠道声明的可跨 Turn 复用投递上下文 JSON（如飞书 thread/root 身份）；一次性 reply/reaction ID 永不写入 | +| `delivery_context_updated_at` | INTEGER | delivery_context 最后更新时间 | `session_turn_usage` 以 `turn_id` 幂等保存已提交 Turn 的 Provider usage，包括累计输入、输出、缓存输入、请求数和最后一次请求的 prompt tokens。它与 Turn 消息批次在同一事务中提交，供 WebUI 状态栏和 `/info` 使用；升级前历史无法可靠回填，因此统计起点以首条 usage 记录为准。 @@ -48,12 +50,14 @@ | `turn_id` | TEXT | 产生该消息的活动 Turn ID | | `iteration` | INTEGER | Agent 工具循环中的迭代序号 | | `completion_status` | TEXT | `completed` / `cancelled` / `interrupted`，旧数据默认 completed | +| `client_visibility` | TEXT | `visible` / `hidden`，默认 visible；hidden 只供模型回放（continuation 内部触发），客户端历史/投影/投递一律过滤 | +| `turn_origin` | TEXT | `user` / `agent_continuation` / `scheduled`，默认 user；客户端据此渲染"后台结果处理"标签而不创建用户气泡 | -`(session_id, seq)` 有唯一索引，防止并发写入重复序号。删除 session 会通过外键级联删除 messages。 +`(session_id, seq)` 有唯一索引，防止并发写入重复序号。删除 session 会通过外键级联删除 messages。索引 `(session_id, client_visibility, seq)` 支撑按可见性分层查询。 -## background_tasks 表 +## background_tasks 表（legacy 兼容，只读过渡） -delegate 后台子任务表。`session_id` 不使用数据库外键，因为 session 使用软删除，关联关系由应用层维护。 +旧 general（无 target）delegate 后台子任务表，由 legacy 适配器写入、`/api/tasks` 只读展示，等待一个版本观察后随旧适配器一起移除。具名 Agent 的后台运行不再写此表。`session_id` 不使用数据库外键，因为 session 使用软删除，关联关系由应用层维护。 | 字段 | 类型 | 说明 | |------|------|------| @@ -72,6 +76,96 @@ delegate 后台子任务表。`session_id` 不使用数据库外键，因为 ses | `finished_at` | INTEGER | 结束时间 | | `created_at` | INTEGER | 创建时间 | +## agent_run_groups 表（schema v6，Agent 编排） + +批量委托的组头。单任务委托不建组；`completion_policy` 决定 background 完成事件形态（当前 background 批量未开放，组仅用于批量 foreground）。 + +| 字段 | 类型 | 说明 | +|------|------|------| +| `id` | TEXT PK | 组 ID | +| `root_session_id` | TEXT | 根会话（软删除，无级联外键，由应用层收敛） | +| `caller_run_id` | TEXT | 发起方 run ID（NULL 表示 Root 发起） | +| `caller_scope_id` | TEXT | 幂等作用域；Root 固定字面量 `"ROOT"` | +| `idempotency_key` | TEXT | 幂等键（当前工具未开放，预留） | +| `mode` | TEXT | foreground / background | +| `completion_policy` | TEXT | all / each | +| `expected_runs` / `terminal_runs` / `abnormal_runs` | INTEGER | 组内 run 计数 | +| `completion_slot_reserved` | INTEGER | 是否预留 background completion 槽 | +| `completion_delivery` / `failure_delivery` | TEXT | 组完成/失败投递 lane（queue/steer） | +| `status` | TEXT | queued / running / completed / partial / failed / timed_out / cancelled / interrupted | +| `deadline_at` / `runtime_generation` / `revision` | INTEGER | 截止、运行代、客户端投影修订号 | +| `created_at` / `updated_at` / `finished_at` | INTEGER | 时间线 | + +## agent_runs 表（schema v6，Agent 编排） + +每次具名委托（foreground 与 background 一致）先落库再执行；`execution_id` 条件更新保证迟到结果丢弃。 + +| 字段 | 类型 | 说明 | +|------|------|------| +| `id` | TEXT PK | run ID | +| `group_id` | TEXT FK | 所属组（RESTRICT） | +| `root_session_id` | TEXT | 根会话 | +| `parent_run_id` | TEXT FK | 父 run（RESTRICT），NULL 表示 Root 直接委托 | +| `caller_agent_id` / `caller_scope_id` | TEXT | 调用方身份；Root 的 caller_scope_id 固定 `"ROOT"` | +| `agent_id` / `definition_hash` / `provider_profile` | TEXT | Definition 快照（绑定运行代，运行中不热切换） | +| `provider_name` / `model_id` | TEXT | Provider 与模型 | +| `mode` | TEXT | foreground / background | +| `depth` | INTEGER | 委托深度（>=1） | +| `plan_item_id` | TEXT | 绑定计划子项（接纳时原子领取） | +| `execution_id` | TEXT | 执行尝试 ID，唯一索引 | +| `task` / `context_json` | TEXT | 任务与调用方上下文 | +| `budget_json` | TEXT | 树级剩余预算 | +| `signal_contract_json` / `signal_delivery` | TEXT | Definition 信号契约快照与投递 lane（queue/steer） | +| `completion_delivery` / `failure_delivery` | TEXT | 完成/失败投递 lane（当前单任务 background 恒为 queue，保留给批量） | +| `status` | TEXT | queued / running / waiting_children / completed / failed / timed_out / cancelled / interrupted | +| `result` / `error` | TEXT | 终态完整结果/错误（get_result 与 tool 结果同源） | +| `prompt_tokens` / `completion_tokens` / `cost` | INTEGER/REAL | Provider usage | +| `tool_calls_count` / `iterations` | INTEGER | 执行统计 | +| `runtime_generation` / `attempt` | INTEGER | 运行代与重试次数 | +| `completion_slot_reserved` | INTEGER | background 完成槽预留 | +| `deadline_at` / `started_at` / `finished_at` / `created_at` / `updated_at` | INTEGER | 时间线 | +| `revision` | INTEGER | 客户端投影修订号 | + +索引：`execution_id` 唯一、`(root_session_id, caller_scope_id, idempotency_key)` 部分唯一、`(root_session_id, created_at DESC)`、`(parent_run_id, created_at)`、`(runtime_generation, status, deadline_at)`（恢复扫描）。 + +## agent_session_state 表（schema v6，Agent 编排） + +每根会话一行，inbox 容量与客户端 revision 的权威计数： + +| 字段 | 说明 | +|------|------| +| `root_session_id` | TEXT PK | +| `revision` | 单调客户端投影修订号 | +| `pending_event_count` | 未消费事件数（容量条件更新） | +| `reserved_completion_slots` | 已接纳 background run 预留的完成槽 | +| `updated_at` | 最后更新时间 | + +容量判断在同一写事务内做条件 `UPDATE`（`pending + reserved + 新增 <= 上限`），杜绝并发 `COUNT(*)` 漂移。 + +## agent_inbox_events 表（schema v6，Agent 编排） + +background 完成/信号投递的唯一事实源：`pending → leased → admitted → consumed`，失败按 token 释放回 pending，超限进 dead-letter，崩溃靠 lease 过期恢复。 + +| 字段 | 说明 | +|------|------| +| `id` | TEXT PK | +| `root_session_id` / `scope_kind` / `scope_id` | 归属（run 或 group，CHECK 互斥） | +| `run_id` / `group_id` | TEXT FK（RESTRICT） | +| `event_type` | signal / completion / group_completion | +| `event_key` | 去重键（signal 含冷却窗口 id） | +| `delivery` | queue / steer | +| `requires_continuation` | 是否反向启动 continuation Turn（cancel 产物为 false） | +| `severity` / `payload_json` | 信号级别与结构化载荷（completion 含 signal_ids） | +| `status` | pending / leased / admitted / consumed / superseded / dead_letter | +| `attempt_count` / `lease_token` / `lease_until` / `next_attempt_at` | 投递尝试与租约 | +| `admitted_turn_id` | steer 事件接纳的 Turn（/stop 按此条件释放） | +| `last_error` | 最近失败原因 | +| `revision` | 投影修订号 | +| `updated_at` | 最后更新时间 | +| `created_at` / `consumed_at` / `superseded_at` / `dead_lettered_at` / `fallback_notified_at` / `fallback_suppressed_reason` | 状态时间线 | + +`(scope_kind, scope_id, event_type, event_key)` 唯一（signal 冷却窗去重）。消费/死信会同步递减 `agent_session_state.pending_event_count`。 + ## task_plans / task_items 表 `task_plans` 保存 session 级任务计划，通过部分唯一索引保证每个 session 最多一个 `status='active'` 的计划。`version` 在任何子项变化时递增，用于 WebSocket 快照排序和乐观并发检查。 diff --git a/src/agent/builtin.rs b/src/agent/builtin.rs new file mode 100644 index 0000000..f0b062d --- /dev/null +++ b/src/agent/builtin.rs @@ -0,0 +1,52 @@ +//! Built-in Agent definitions, released to the user config directory on +//! first run just like built-in skills. A released definition is a regular +//! user-editable file afterwards; the installer never overwrites it. + +use std::path::Path; + +use crate::config::LLMProviderConfig; + +mod embedded { + include!(concat!(env!("OUT_DIR"), "/embedded_agents.rs")); +} + +/// Install built-in Agent definitions into `/agents/`. Files +/// that already exist (user-modified or user-created) are left untouched. +pub fn install_builtin_agents(config_dir: &Path, profiles: &std::collections::HashMap) { + let agents_dir = config_dir.join("agents"); + if let Err(error) = std::fs::create_dir_all(&agents_dir) { + tracing::warn!(dir = %agents_dir.display(), error = %error, "Failed to create agents directory"); + return; + } + for agent in embedded::EMBEDDED_AGENTS { + let path = agents_dir.join(format!("{}.md", agent.name)); + if path.exists() { + continue; + } + if let Err(error) = std::fs::write(&path, agent.content) { + tracing::warn!(name = agent.name, error = %error, "Failed to install built-in Agent definition"); + continue; + } + let profile = match profiles.get("default").cloned() { + Some(profile) => profile, + None => { + tracing::warn!( + name = agent.name, + "Skipping built-in Agent validation: no 'default' provider profile configured" + ); + continue; + } + }; + // Validate the released file immediately so a future catalog load + // cannot fail on a broken built-in. Validation failure keeps the + // file (the user can edit it) but logs loudly. + match super::definition::parse_definition(&path, std::sync::Arc::new(profile)) { + Ok(_) => { + tracing::info!(name = agent.name, dir = %path.display(), "Installed built-in Agent definition"); + } + Err(error) => { + tracing::warn!(name = agent.name, error = %error, "Installed built-in Agent definition failed validation"); + } + } + } +} diff --git a/src/agent/coordinator.rs b/src/agent/coordinator.rs index 5ca66ce..102eda6 100644 --- a/src/agent/coordinator.rs +++ b/src/agent/coordinator.rs @@ -1125,17 +1125,13 @@ mod tests { ) -> (Arc, tempfile::TempDir) { let dir = tempfile::tempdir().unwrap(); let storage = Arc::new(Storage::new(&dir.path().join("coord.db")).await.unwrap()); - let (notify_tx, _notify_rx) = tokio::sync::mpsc::unbounded_channel(); let catalog = write_catalog(dir.path()); let manager = Arc::new( SubAgentManager::new( provider_config(), Arc::new(ToolRegistry::new()), Some(storage.clone()), - notify_tx, - 1, None, - crate::task_supervisor::TaskSupervisor::new(), ) .with_catalog(Arc::new(catalog)), ); @@ -1627,17 +1623,13 @@ mod tests { let coordinator2 = { let dir = tempfile::tempdir().unwrap(); let storage2 = Arc::new(Storage::new(&dir.path().join("c2.db")).await.unwrap()); - let (notify_tx, _) = tokio::sync::mpsc::unbounded_channel(); let catalog = write_catalog(dir.path()); let manager = Arc::new( SubAgentManager::new( provider_config(), Arc::new(ToolRegistry::new()), Some(storage2.clone()), - notify_tx, - 1, None, - crate::task_supervisor::TaskSupervisor::new(), ) .with_catalog(Arc::new(catalog)), ); diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 961ae5b..3b56d27 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -1,4 +1,5 @@ pub mod agent_loop; +pub mod builtin; pub mod catalog; pub mod context_compressor; pub mod coordinator; @@ -24,8 +25,7 @@ pub use projection::AgentProjectionHub; pub use run::{AgentBudget, AgentExecutionContext}; pub use steering::{SteeringDrain, SteeringPushError, TurnInput, TurnInputSource, TurnMailbox}; pub use sub_agent::{ - DelegateContext, ExecutionMode, SubAgentConfig, SubAgentError, SubAgentManager, SubAgentResult, - TaskNotification, TaskStatus, + ExecutionMode, SubAgentConfig, SubAgentError, SubAgentManager, SubAgentResult, TaskStatus, }; pub use system_prompt::{ PromptContext, PromptSection, SystemPromptBuilder, build_sub_agent_system_prompt, diff --git a/src/agent/sub_agent.rs b/src/agent/sub_agent.rs index ff78e1a..a15377e 100644 --- a/src/agent/sub_agent.rs +++ b/src/agent/sub_agent.rs @@ -1,11 +1,6 @@ -use std::collections::HashSet; use std::sync::Arc; use std::time::Instant; -use dashmap::DashMap; -use tokio::sync::Semaphore; -use tokio_util::sync::CancellationToken; -use uuid::Uuid; use crate::agent::AgentError; use crate::agent::AgentLoop; @@ -16,29 +11,7 @@ use crate::providers::{LLMProvider, create_provider}; use crate::skills::SkillsLoader; use crate::tools::{ToolExecutionContext, ToolRegistry}; -tokio::task_local! { - pub(crate) static DELEGATE_CONTEXT: DelegateContext; -} - -/// Read the delegate context from the current task. Returns an error if not set. -pub fn get_delegate_context() -> Result { - DELEGATE_CONTEXT - .try_with(|ctx| ctx.clone()) - .map_err(|_| "DELEGATE_CONTEXT not set".to_string()) -} - const DEFAULT_MAX_ITERATIONS: usize = 99; -const DEFAULT_TIMEOUT_SECS: u64 = 3600; -const MAX_INLINE_RESULT_CHARS: usize = 8000; - -const DEFAULT_READONLY_TOOLS: &[&str] = &[ - "file_read", - "file_search", - "content_search", - "web_fetch", - "http_request", - "calculator", -]; #[derive(Debug, Clone)] pub struct SubAgentConfig { @@ -81,26 +54,8 @@ pub enum TaskStatus { TimedOut, } -#[derive(Debug, Clone)] -pub struct TaskNotification { - pub task_id: String, - pub session_id: String, - pub channel: String, - pub chat_id: String, - pub status: TaskStatus, - pub result_summary: String, -} - -#[derive(Debug, Clone)] -pub struct DelegateContext { - pub session_id: String, - pub channel: String, - pub chat_id: String, -} - #[derive(Debug)] pub enum SubAgentError { - TooManyTasks(usize), ProviderCreation(String), Storage(String), Other(String), @@ -109,7 +64,6 @@ pub enum SubAgentError { impl std::fmt::Display for SubAgentError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::TooManyTasks(max) => write!(f, "后台任务已达上限({})，请稍后重试", max), Self::ProviderCreation(e) => write!(f, "provider creation failed: {}", e), Self::Storage(e) => write!(f, "storage error: {}", e), Self::Other(e) => write!(f, "{}", e), @@ -123,14 +77,8 @@ pub struct SubAgentManager { provider_config: LLMProviderConfig, full_tools: Arc, storage: Option>, - active_tasks: Arc>, - background_permits: Arc, - notify_tx: tokio::sync::mpsc::UnboundedSender, - max_concurrent_background_tasks: usize, skills_loader: Option>, work_manager: Option>, - task_supervisor: crate::task_supervisor::TaskSupervisor, - admission: crate::gateway::reload::RuntimeAdmission, catalog: Arc, execution_gate: Arc, /// Late-bound durable Coordinator. Signals are only available to runs @@ -164,23 +112,14 @@ impl SubAgentManager { provider_config: LLMProviderConfig, full_tools: Arc, storage: Option>, - notify_tx: tokio::sync::mpsc::UnboundedSender, - max_concurrent_background_tasks: usize, skills_loader: Option>, - task_supervisor: crate::task_supervisor::TaskSupervisor, ) -> Self { Self { provider_config, full_tools, storage, - active_tasks: Arc::new(DashMap::new()), - background_permits: Arc::new(Semaphore::new(max_concurrent_background_tasks)), - notify_tx, - max_concurrent_background_tasks, skills_loader, work_manager: None, - task_supervisor, - admission: crate::gateway::reload::RuntimeAdmission::open(), catalog: Arc::new(crate::agent::AgentCatalog::legacy()), execution_gate: crate::agent::gate::ExecutionGate::unbounded(), coordinator: std::sync::RwLock::new(None), @@ -202,14 +141,6 @@ impl SubAgentManager { .and_then(std::sync::Weak::upgrade) } - pub(crate) fn with_admission( - mut self, - admission: crate::gateway::reload::RuntimeAdmission, - ) -> Self { - self.admission = admission; - self - } - pub fn with_catalog(mut self, catalog: Arc) -> Self { self.catalog = catalog; self @@ -232,32 +163,6 @@ impl SubAgentManager { 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(), - None => DEFAULT_READONLY_TOOLS.iter().copied().collect(), - }; - let filtered = ToolRegistry::new(); - for (name, tool) in self.full_tools.iter() { - if allowed_set.contains(name.as_str()) - && tool.delegation_policy() == crate::tools::DelegationPolicy::Delegatable - { - filtered.register_raw(name, tool); - } - } - Arc::new(filtered) - } - - fn get_skills_prompt(&self, tools: &ToolRegistry) -> Option { - let has_get_skill = tools.iter().iter().any(|(name, _)| name == "get_skill"); - if has_get_skill && let Some(ref loader) = self.skills_loader { - let prompt = loader.build_skills_prompt(); - if !prompt.is_empty() { - return Some(prompt); - } - } - None - } pub(crate) fn resolve_agent( &self, @@ -266,38 +171,10 @@ impl SubAgentManager { task_id: &str, ) -> Result { let Some(target) = config.target.as_deref() else { - if caller.agent.is_some() { - return Err(SubAgentError::Other( - "named child Agents cannot use the legacy general Agent".to_string(), - )); - } - let browser_session_id = config - .session_id - .clone() - .or_else(|| caller.session_id.clone()) - .or_else(|| { - get_delegate_context() - .ok() - .map(|context| context.session_id) - }) - .unwrap_or_else(|| format!("sub-agent:{task_id}")); - let tools = self.filter_tools(&config.allowed_tools); - return Ok(ResolvedAgentRun { - provider_config: Arc::new(self.provider_config.clone()), - skills_prompt: self.get_skills_prompt(&tools), - tools, - timeout_secs: config.timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS), - max_iterations: config.max_iterations.unwrap_or(DEFAULT_MAX_ITERATIONS), - max_result_chars: MAX_INLINE_RESULT_CHARS, - role_prompt: None, - tool_context: ToolExecutionContext::for_session(browser_session_id) - .with_cancellation(caller.cancellation.child_token()) - .with_execution_gate(self.execution_gate.clone()), - agent_id: None, - definition_hash: None, - llm_profile: None, - signal_contract: None, - }); + return Err(SubAgentError::Other( + "delegate requires a named target Agent; the legacy anonymous general Agent is no longer available" + .to_string(), + )); }; if !self.catalog.enabled() { @@ -526,33 +403,8 @@ impl SubAgentManager { Ok(agent) } - pub async fn run_inline( - &self, - config: SubAgentConfig, - ) -> Result { - let mut caller = ToolExecutionContext::default(); - if let Some(session_id) = config.session_id.clone() { - caller.session_id = Some(session_id); - } - self.run_foreground(config, &caller).await - } - - pub async fn run_foreground( - &self, - config: SubAgentConfig, - caller: &ToolExecutionContext, - ) -> Result { - let task_id = generate_task_id(); - let resolved = self.resolve_agent(&config, caller, &task_id)?; - self.assign_work_item(&config, &task_id).await?; - let result = self.execute_resolved(&config, resolved, &task_id).await?; - self.finish_work_item(&config, &result).await; - Ok(result) - } - - /// Execute an already-resolved Agent without any plan-item side effects. - /// The durable Coordinator owns plan admission/terminal updates itself; - /// the legacy path wraps this with assign/finish hooks. + /// Execute an already-resolved Agent. The durable Coordinator owns run + /// admission and terminal commits; this is the shared execution core. pub(crate) async fn execute_resolved( &self, config: &SubAgentConfig, @@ -668,512 +520,6 @@ impl SubAgentManager { }, }) } - - pub async fn run_foreground_batch( - &self, - configs: Vec, - caller: &ToolExecutionContext, - ) -> Result, SubAgentError> { - if configs.is_empty() { - return Err(SubAgentError::Other( - "foreground batch must contain at least one task".to_string(), - )); - } - if let Some(parent) = caller.agent.as_ref() { - let definition = self.catalog.get(&parent.current_agent_id).ok_or_else(|| { - SubAgentError::Other(format!( - "caller Agent '{}' is not present in the active catalog", - parent.current_agent_id - )) - })?; - if configs.len() > definition.limits.max_children { - return Err(SubAgentError::Other(format!( - "Agent '{}' may create at most {} children per delegate call", - parent.current_agent_id, definition.limits.max_children - ))); - } - if configs.len() > parent.budget.remaining_runs { - return Err(SubAgentError::Other( - "delegation run budget is smaller than the requested batch".to_string(), - )); - } - if configs.len() > parent.remaining_tree_runs(self.catalog.max_runs_per_tree()) { - return Err(SubAgentError::Other( - "delegation tree capacity is smaller than the requested batch".to_string(), - )); - } - } else if self.catalog.enabled() && configs.len() > self.catalog.max_runs_per_tree() { - return Err(SubAgentError::Other(format!( - "ROOT batch exceeds max_runs_per_tree ({})", - self.catalog.max_runs_per_tree() - ))); - } - let futures: Vec<_> = configs - .into_iter() - .map(|config| { - let caller = caller.clone(); - async move { self.run_foreground(config, &caller).await } - }) - .collect(); - - let results = futures_util::future::join_all(futures).await; - Ok(results - .into_iter() - .enumerate() - .map(|(index, result)| { - result.unwrap_or_else(|error| SubAgentResult { - task_id: format!("rejected-{}-{}", index + 1, generate_task_id()), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::Failed(error.to_string()), - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }) - }) - .collect()) - } - - pub async fn run_background( - &self, - config: SubAgentConfig, - ctx: DelegateContext, - ) -> Result { - let activity = self.admission.try_enter().ok_or_else(|| { - SubAgentError::Other( - "gateway is draining for configuration reload and cannot accept background tasks" - .to_string(), - ) - })?; - let permit = self - .background_permits - .clone() - .try_acquire_owned() - .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(); - - // Write DB: pending - if let Some(ref storage) = self.storage { - let allowed_tools_json = config - .allowed_tools - .as_ref() - .and_then(|v| serde_json::to_string(v).ok()); - let record = crate::storage::BackgroundTask { - id: task_id.clone(), - session_id: ctx.session_id.clone(), - channel: ctx.channel.clone(), - chat_id: ctx.chat_id.clone(), - prompt: config.prompt.clone(), - allowed_tools: allowed_tools_json, - status: "pending".to_string(), - result: None, - error: None, - tool_calls_count: 0, - iterations: 0, - started_at: None, - finished_at: None, - created_at: chrono::Utc::now().timestamp_millis(), - }; - storage - .create_background_task(&record) - .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); - let timeout_human = format_duration(timeout_secs); - let skills_prompt = self.get_skills_prompt(&tools); - let system_prompt = build_sub_agent_system_prompt( - &config.prompt, - &timeout_human, - &tools, - &self.provider_config.workspace_dir, - &self.provider_config.model_id, - skills_prompt, - ); - let provider_config = self.provider_config.clone(); - let storage = self.storage.clone(); - let notify_tx = self.notify_tx.clone(); - let active_tasks = Arc::clone(&self.active_tasks); - let shutdown = self.task_supervisor.cancellation_token(); - let execution_gate = self.execution_gate.clone(); - - let tid = task_id.clone(); - let sess_id = ctx.session_id.clone(); - 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}"), - async move { - let _activity = activity; - let _permit = permit; - let started_at = chrono::Utc::now().timestamp_millis(); - - // Update DB: running - if let Some(ref s) = storage { - let _ = s - .update_background_task_status(&tid, crate::storage::background_task::BackgroundTaskUpdate { - status: "running", - result: None, - error: None, - started_at: Some(started_at), - finished_at: None, - tool_calls_count: None, - iterations: None, - }) - .await; - } - - let mut provider = create_provider(provider_config.clone()).ok(); - if let Some(ref mut p) = provider - && let Some(ref s) = storage - { - p.set_storage(s.clone()); - } - let provider_result: Option> = provider.map(Arc::from); - - let result = match provider_result { - Some(provider) => { - let agent = AgentLoop::with_provider_and_tools( - provider, - tools, - DEFAULT_MAX_ITERATIONS, - provider_config.model_id.clone(), - provider_config.workspace_dir.clone(), - provider_config.input_types.clone(), - ) - .with_context_window(provider_config.token_limit); - - let history = vec![ - ChatMessage::system(system_prompt), - ChatMessage::user(&prompt), - ]; - - let tool_context = ToolExecutionContext::for_session(&sess_id) - .with_cancellation(cancel_token.clone()) - .with_execution_gate(execution_gate.clone()); - tokio::select! { - r = tokio::time::timeout( - std::time::Duration::from_secs(timeout_secs), - agent.process_with_context(history, tool_context), - ) => { - match r { - Ok(Ok(agent_result)) => { - let tool_calls_count = agent_result.emitted_messages - .iter().filter(|m| m.tool_calls.is_some()).count(); - let iterations = agent_result.emitted_messages - .iter().filter(|m| m.role == "assistant" && m.tool_calls.is_some()).count(); - SubAgentResult { - task_id: tid.clone(), - content: agent_result.final_response.content.clone(), - content_truncated: false, - full_content: agent_result.final_response.content, - status: TaskStatus::Completed, - tool_calls_count, - iterations, - duration_ms: 0, - } - }, - Ok(Err(error)) => SubAgentResult { - task_id: tid.clone(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: terminal_status_from_error(error), - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }, - Err(_) => SubAgentResult { - task_id: tid.clone(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::TimedOut, - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }, - } - } - _ = cancel_token.cancelled() => SubAgentResult { - task_id: tid.clone(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::Cancelled, - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }, - _ = shutdown.cancelled() => SubAgentResult { - task_id: tid.clone(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::Cancelled, - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }, - } - } - None => SubAgentResult { - task_id: tid.clone(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::Failed("provider creation failed".into()), - tool_calls_count: 0, - iterations: 0, - duration_ms: 0, - }, - }; - - let finished_at = chrono::Utc::now().timestamp_millis(); - let duration_ms = (finished_at - started_at) as u64; - - let (status_str, error_val) = match &result.status { - TaskStatus::Completed => ("completed".to_string(), None), - TaskStatus::Failed(e) => ("failed".to_string(), Some(e.clone())), - TaskStatus::Cancelled => ("cancelled".to_string(), None), - 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 { - status: &status_str, - result: Some(&result.content), - error: error_val.as_deref(), - started_at: Some(started_at), - finished_at: Some(finished_at), - tool_calls_count: Some(result.tool_calls_count as i64), - iterations: Some(result.iterations as i64), - }) - .await; - } - - let _ = notify_tx.send(TaskNotification { - task_id: tid.clone(), - session_id: sess_id, - channel: ch, - chat_id: cid, - status: result.status, - result_summary: summarize_for_notification(&result.content, duration_ms), - }); - - active_tasks.remove(&tid); - }); - - if !spawned { - self.active_tasks.remove(&task_id); - 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("gateway shutdown"), - started_at: None, - finished_at: Some(chrono::Utc::now().timestamp_millis()), - tool_calls_count: None, - iterations: None, - }, - ) - .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(), - )); - } - - Ok(task_id) - } - - pub async fn cancel_task(&self, task_id: &str) -> Result { - if let Some((_, token)) = self.active_tasks.remove(task_id) { - token.cancel(); - if let Some(ref s) = self.storage { - s.update_background_task_status( - task_id, - crate::storage::background_task::BackgroundTaskUpdate { - status: "cancelled", - result: None, - error: None, - started_at: None, - finished_at: Some(chrono::Utc::now().timestamp_millis()), - tool_calls_count: None, - iterations: None, - }, - ) - .await - .map_err(|e| SubAgentError::Storage(e.to_string()))?; - } - Ok(true) - } else if let Some(ref s) = self.storage { - match s.get_background_task(task_id).await { - Ok(task) => match task.status.as_str() { - "pending" | "running" => { - tracing::warn!(task_id, "task in DB but not in active_tasks"); - Ok(false) - } - _ => Ok(false), - }, - Err(_) => Ok(false), - } - } else { - Ok(false) - } - } - - pub async fn check_task(&self, task_id: &str) -> Option { - if let Some(ref s) = self.storage { - s.get_background_task(task_id).await.ok() - } else { - None - } - } - - pub async fn list_tasks(&self, session_id: &str) -> Vec { - if let Some(ref s) = self.storage { - s.list_background_tasks(session_id) - .await - .unwrap_or_default() - } else { - vec![] - } - } - - pub async fn cancel_by_session(&self, session_id: &str) { - // Cancel all running tasks for a session by checking DB - if let Some(ref s) = self.storage - && let Ok(tasks) = s.list_background_tasks(session_id).await - { - for task in &tasks { - if task.status == "pending" || task.status == "running" { - let _ = self.cancel_task(&task.id).await; - } - } - } - } - - 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 terminal_status_from_error(error: AgentError) -> TaskStatus { @@ -1184,9 +530,6 @@ fn terminal_status_from_error(error: AgentError) -> TaskStatus { } } -fn generate_task_id() -> String { - Uuid::new_v4().to_string() -} fn format_duration(seconds: u64) -> String { if seconds < 60 { @@ -1214,23 +557,12 @@ fn truncate_sub_agent_result_at(content: &str, max_chars: usize) -> (String, boo } } -fn summarize_for_notification(content: &str, _duration_ms: u64) -> String { - const MAX_SUMMARY_BYTES: usize = 500; - if content.len() <= MAX_SUMMARY_BYTES { - content.to_string() - } else { - let truncate_at = content.floor_char_boundary(MAX_SUMMARY_BYTES); - format!("{}...", &content[..truncate_at]) - } -} - #[cfg(test)] mod tests { use super::*; use std::collections::HashMap; - fn manager(max_tasks: usize) -> SubAgentManager { - let (notify_tx, _notify_rx) = tokio::sync::mpsc::unbounded_channel(); + fn manager() -> SubAgentManager { SubAgentManager::new( LLMProviderConfig { provider_type: "openai".into(), @@ -1251,179 +583,14 @@ mod tests { }, Arc::new(ToolRegistry::new()), None, - notify_tx, - max_tasks, None, - crate::task_supervisor::TaskSupervisor::new(), ) } - #[tokio::test] - async fn background_limit_is_enforced_by_atomic_permit() { - let manager = manager(1); - let _permit = manager - .background_permits - .clone() - .try_acquire_owned() - .unwrap(); - let error = manager - .run_background( - SubAgentConfig { - target: None, - prompt: "test".into(), - context: None, - mode: ExecutionMode::Background, - allowed_tools: None, - max_iterations: None, - timeout_secs: Some(1), - plan_item_id: None, - session_id: None, - }, - DelegateContext { - session_id: "cli:test:dialog".into(), - channel: "cli".into(), - chat_id: "test".into(), - }, - ) - .await - .unwrap_err(); - - assert!(matches!(error, SubAgentError::TooManyTasks(1))); - } - - #[test] - fn reload_tool_is_never_delegated_to_sub_agents() { - let manager = manager(1); - manager - .full_tools - .register(crate::tools::ReloadConfigTool::new( - crate::gateway::reload::ReloadHandle::unavailable(), - )); - - let filtered = manager.filter_tools(&Some(vec!["reload_config".to_string()])); - assert!(filtered.get("reload_config").is_none()); - } - - fn catalog_with_agents( - root: &std::path::Path, - max_runs_per_tree: usize, - ) -> crate::agent::AgentCatalog { - std::fs::create_dir_all(root.join("agents")).unwrap(); - let write = |id: &str, delegates: &[&str]| { - let delegates = (!delegates.is_empty()).then(|| { - format!( - "delegates:\n{}\n", - delegates - .iter() - .map(|name| format!(" - {name}")) - .collect::>() - .join("\n") - ) - }); - std::fs::write( - root.join("agents").join(format!("{id}.md")), - format!( - "---\nid: {id}\ndescription: {id} role\nllm_profile: research\n{}---\n# Role\n\nDo the assigned work.\n", - delegates.unwrap_or_default() - ), - ) - .unwrap(); - }; - write("researcher", &["reviewer"]); - write("reviewer", &[]); - let tools = ToolRegistry::new(); - let loader = crate::skills::SkillsLoader::new_for_testing( - root.join("skills"), - root.join("external-skills"), - ); - let profiles = HashMap::from([( - "research".to_string(), - LLMProviderConfig { - provider_type: "openai".into(), - name: "test".into(), - base_url: "http://localhost".into(), - api_key: "test".into(), - extra_headers: HashMap::new(), - model_id: "test".into(), - temperature: None, - max_tokens: None, - model_extra: HashMap::new(), - max_tool_iterations: 1, - token_limit: 4096, - workspace_dir: std::env::temp_dir(), - input_types: vec!["text".into()], - price_input_per_million: None, - price_output_per_million: None, - }, - )]); - let config = crate::config::AgentOrchestrationConfig { - enabled: true, - definitions_dir: "agents".to_string(), - root_delegates: vec!["researcher".to_string()], - max_runs_per_tree, - ..Default::default() - }; - crate::agent::AgentCatalog::load(&config, root, &profiles, &tools, &loader, 1).unwrap() - } - - #[tokio::test] - async fn foreground_batch_rejects_when_tree_capacity_exhausted() { - let root = tempfile::tempdir().unwrap(); - let catalog = catalog_with_agents(root.path(), 2); - let manager = manager(1).with_catalog(Arc::new(catalog)); - - let caller_context = Arc::new(crate::agent::AgentExecutionContext { - root_session_id: "cli:test:dialog".to_string(), - root_turn_id: None, - run_id: "run-root".to_string(), - execution_id: "run-root".to_string(), - group_id: None, - parent_run_id: None, - caller_agent_id: "ROOT".to_string(), - current_agent_id: "researcher".to_string(), - ancestry: vec!["researcher".to_string()], - depth: 1, - plan_item_id: None, - cancellation: CancellationToken::new(), - budget: crate::agent::AgentBudget { - remaining_runs: 15, - remaining_depth: 3, - }, - tree_runs: Arc::new(std::sync::atomic::AtomicUsize::new(2)), - signal_contract: None, - emitted_signals: Arc::new(std::sync::Mutex::new(Vec::new())), - }); - let caller = - ToolExecutionContext::for_session("cli:test:dialog").with_agent(caller_context); - - let error = manager - .run_foreground_batch( - vec![SubAgentConfig { - target: Some("reviewer".to_string()), - prompt: "test".to_string(), - context: None, - mode: ExecutionMode::Foreground, - allowed_tools: None, - max_iterations: None, - timeout_secs: None, - plan_item_id: None, - session_id: Some("cli:test:dialog".to_string()), - }], - &caller, - ) - .await - .unwrap_err(); - assert!( - matches!(error, SubAgentError::Other(message) if message.contains("tree capacity")) - ); - } - - #[tokio::test] - async fn foreground_batch_preserves_per_task_rejections() { - let manager = manager(1); - let config = |target: &str| SubAgentConfig { - target: Some(target.to_string()), - prompt: "test".to_string(), + fn config(target: Option<&str>) -> SubAgentConfig { + SubAgentConfig { + target: target.map(str::to_string), + prompt: "test".into(), context: None, mode: ExecutionMode::Foreground, allowed_tools: None, @@ -1431,21 +598,37 @@ mod tests { timeout_secs: None, plan_item_id: None, session_id: Some("cli:test:dialog".to_string()), - }; + } + } - let results = manager - .run_foreground_batch( - vec![config("missing-a"), config("missing-b")], - &ToolExecutionContext::for_session("cli:test:dialog"), - ) - .await - .unwrap(); - - assert_eq!(results.len(), 2); - assert!( - results - .iter() - .all(|result| matches!(result.status, TaskStatus::Failed(_))) + #[test] + fn reload_tool_is_never_delegated_to_sub_agents() { + let tool = crate::tools::ReloadConfigTool::new( + crate::gateway::reload::ReloadHandle::unavailable(), + ); + assert_eq!( + crate::tools::Tool::delegation_policy(&tool), + crate::tools::DelegationPolicy::RootOnly ); } + + #[test] + fn resolve_agent_rejects_missing_target() { + let manager = manager(); + let error = match manager.resolve_agent(&config(None), &ToolExecutionContext::default(), "t-1") { + Ok(_) => panic!("expected rejection"), + Err(error) => error, + }; + assert!(matches!(error, SubAgentError::Other(message) if message.contains("named target"))); + } + + #[test] + fn resolve_agent_rejects_unknown_target_without_catalog() { + let manager = manager(); + let error = match manager.resolve_agent(&config(Some("ghost")), &ToolExecutionContext::default(), "t-2") { + Ok(_) => panic!("expected rejection"), + Err(error) => error, + }; + assert!(matches!(error, SubAgentError::Other(message) if message.contains("orchestration"))); + } } diff --git a/src/config/mod.rs b/src/config/mod.rs index afc1c92..4af79a4 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -204,7 +204,10 @@ impl Default for AgentOrchestrationConfig { Self { enabled: false, definitions_dir: "agents".to_string(), - root_delegates: Vec::new(), + // The built-in general-purpose Agent is released automatically; + // it is delegated by default so orchestration works out of the + // box. Explicit configuration fully overrides this list. + root_delegates: vec!["general-purpose".to_string()], max_tree_depth: 4, max_runs_per_tree: 16, max_concurrent_runs: 6, diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 929f449..b6a5ab4 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -210,7 +210,6 @@ impl GatewayState { .with_admission(admission.clone()), browser_config, health, - config.gateway.max_concurrent_background_tasks, )?; let session_manager = Arc::new(session_manager); session_manager.bind_inbox_wake(); diff --git a/src/session/session.rs b/src/session/session.rs index 870d94e..0a5f4e0 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -1718,7 +1718,6 @@ pub struct SessionManager { pub(super) bus: Arc, memory_manager: Arc, work_manager: Arc, - sub_agent_manager: Arc, agent_catalog: Arc, execution_gate: Arc, agent_coordinator: Option>, @@ -1929,7 +1928,6 @@ impl SessionManager { services: SessionManagerServices, browser_config: Option, health: Arc, - max_concurrent_background_tasks: usize, ) -> Result { let SessionManagerServices { bus, @@ -1958,6 +1956,12 @@ impl SessionManager { .map_err(|error| AgentError::Other(format!("failed to create tools: {error}")))?, ); + // Release built-in Agent definitions (like built-in skills) before + // the catalog loads; existing user files are never overwritten. + crate::agent::builtin::install_builtin_agents( + &catalog_preparation.config_dir, + &catalog_preparation.provider_profiles, + ); let agent_catalog = Arc::new( crate::agent::AgentCatalog::load( &catalog_preparation.config, @@ -1976,18 +1980,13 @@ impl SessionManager { }; // 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(), ) - .with_admission(admission.clone()) .with_catalog(agent_catalog.clone()) .with_execution_gate(execution_gate.clone()) .with_work_manager(work_manager.clone()), @@ -2018,29 +2017,6 @@ impl SessionManager { tools.register(delegate_tool); tools.register(crate::tools::ReloadConfigTool::new(reload.clone())); - // Start background task notification consumer - let sm_bus = bus.clone(); - task_supervisor.spawn("background-task-notifications", async move { - 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, - delivery: None, - }; - let _ = sm_bus.publish_outbound(outbound).await; - } - }); - // Start periodic background task cleanup (every hour, TTL 24h) let cleanup_storage = storage.clone(); task_supervisor.spawn("background-task-cleanup", async move { @@ -2072,7 +2048,6 @@ impl SessionManager { bus, memory_manager, work_manager, - sub_agent_manager, agent_catalog, execution_gate, agent_coordinator, @@ -2447,9 +2422,6 @@ impl SessionManager { // after releasing the session lock. Named durable runs are // cancelled with suppress_continuation so no continuation Turn // restarts after an explicit stop. - self.sub_agent_manager - .cancel_by_session(&sid.to_string()) - .await; if let Some(coordinator) = self.agent_coordinator.as_ref() && let Err(error) = coordinator .cancel_session(&sid.to_string(), "stopped by user") @@ -4048,18 +4020,13 @@ fn spawn_agent_worker( if let Some(handle) = wakeup_handle { tool_context = tool_context.with_turn_wakeup(handle); } - let process_result = crate::agent::sub_agent::DELEGATE_CONTEXT.scope( - crate::agent::DelegateContext { - session_id: unified_str2, - channel: chan2.clone(), - chat_id: cid2.clone(), - }, - agent.process_streaming_with_context( + let process_result = agent + .process_streaming_with_context( history_out.clone(), agent_turn.clone(), tool_context.clone(), - ), - ).await; + ) + .await; let mut result = match process_result { Ok(r) => r, Err(AgentError::LlmError(ref msg)) @@ -5081,24 +5048,6 @@ impl SessionManager { } } -fn format_task_notification( - task_id: &str, - status: &crate::agent::TaskStatus, - summary: &str, -) -> String { - match status { - crate::agent::TaskStatus::Completed => format!( - "📋 后台任务完成\n\n任务 ID: {}\n\n结果:\n{}", - task_id, summary - ), - crate::agent::TaskStatus::Failed(err) => { - format!("📋 后台任务失败\n\n任务 ID: {}\n错误: {}", task_id, err) - } - crate::agent::TaskStatus::Cancelled => format!("📋 后台任务已取消\n\n任务 ID: {}", task_id), - crate::agent::TaskStatus::TimedOut => format!("📋 后台任务超时\n\n任务 ID: {}", task_id), - } -} - #[cfg(test)] mod slash_command_tests { use super::{ diff --git a/src/storage/background_task.rs b/src/storage/background_task.rs index 669ee04..a01d1ed 100644 --- a/src/storage/background_task.rs +++ b/src/storage/background_task.rs @@ -17,13 +17,3 @@ pub struct BackgroundTask { pub finished_at: Option, pub created_at: i64, } - -pub(crate) struct BackgroundTaskUpdate<'a> { - pub status: &'a str, - pub result: Option<&'a str>, - pub error: Option<&'a str>, - pub started_at: Option, - pub finished_at: Option, - pub tool_calls_count: Option, - pub iterations: Option, -} diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 0020ca1..a0f1f67 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -1490,33 +1490,6 @@ impl Storage { Ok(()) } - pub(crate) async fn update_background_task_status( - &self, - id: &str, - update: crate::storage::background_task::BackgroundTaskUpdate<'_>, - ) -> Result<(), StorageError> { - sqlx::query( - r#" - UPDATE background_tasks - SET status = ?, result = COALESCE(?, result), error = COALESCE(?, error), - started_at = COALESCE(?, started_at), finished_at = COALESCE(?, finished_at), - tool_calls_count = COALESCE(?, tool_calls_count), - iterations = COALESCE(?, iterations) - WHERE id = ? - "#, - ) - .bind(update.status) - .bind(update.result) - .bind(update.error) - .bind(update.started_at) - .bind(update.finished_at) - .bind(update.tool_calls_count) - .bind(update.iterations) - .bind(id) - .execute(self.pool()) - .await?; - Ok(()) - } pub async fn get_background_task( &self, @@ -1823,48 +1796,6 @@ mod tests { assert_eq!(sentinel_count, 1); } - #[tokio::test] - async fn background_task_completion_persists_execution_metrics() { - let (storage, _dir) = create_test_storage().await; - let task = crate::storage::BackgroundTask { - id: "task-metrics".into(), - session_id: "cli:test:dialog".into(), - channel: "cli".into(), - chat_id: "test".into(), - prompt: "measure".into(), - allowed_tools: None, - status: "pending".into(), - result: None, - error: None, - tool_calls_count: 0, - iterations: 0, - started_at: None, - finished_at: None, - created_at: 1, - }; - storage.create_background_task(&task).await.unwrap(); - - storage - .update_background_task_status( - &task.id, - crate::storage::background_task::BackgroundTaskUpdate { - status: "completed", - result: Some("done"), - error: None, - started_at: Some(2), - finished_at: Some(3), - tool_calls_count: Some(4), - iterations: Some(5), - }, - ) - .await - .unwrap(); - - let persisted = storage.get_background_task(&task.id).await.unwrap(); - assert_eq!(persisted.tool_calls_count, 4); - assert_eq!(persisted.iterations, 5); - } - #[tokio::test] async fn webui_lists_recent_tasks_across_sessions() { let (storage, _dir) = create_test_storage().await; diff --git a/src/tools/bash.rs b/src/tools/bash.rs index 74f6d89..98ddcd5 100644 --- a/src/tools/bash.rs +++ b/src/tools/bash.rs @@ -86,6 +86,10 @@ impl Tool for BashTool { "bash" } + fn delegation_policy(&self) -> crate::tools::DelegationPolicy { + crate::tools::DelegationPolicy::Delegatable + } + fn description(&self) -> &str { "Execute a bash shell command and return its output. Use with caution." } diff --git a/src/tools/delegate.rs b/src/tools/delegate.rs index 71fa2db..a73cffc 100644 --- a/src/tools/delegate.rs +++ b/src/tools/delegate.rs @@ -87,7 +87,7 @@ impl DelegateTool { .collect(); let mut target = json!({ "type": "string", - "description": "目标 Agent ID。省略时仅使用兼容 general Agent" + "description": "目标具名 Agent ID（来自 agent_orchestration 的 agents 目录）" }); if !targets.is_empty() { target["enum"] = json!(targets); @@ -155,20 +155,9 @@ impl DelegateTool { args: &Value, context: &ToolExecutionContext, ) -> anyhow::Result { - let requested_mode = args - .get("mode") - .and_then(Value::as_str) - .unwrap_or("foreground"); - if matches!(requested_mode, "inline" | "parallel") { - tracing::warn!( - mode = requested_mode, - "deprecated delegate mode used; migrate to foreground with an optional tasks array" - ); - } - let (mode, legacy_parallel) = match requested_mode { - "foreground" | "inline" => (ExecutionMode::Foreground, false), - "background" => (ExecutionMode::Background, false), - "parallel" => (ExecutionMode::Foreground, true), + let mode = match args.get("mode").and_then(Value::as_str).unwrap_or("foreground") { + "foreground" => ExecutionMode::Foreground, + "background" => ExecutionMode::Background, other => { return Ok(failure(format!( "unknown mode '{other}'; supported modes are foreground and background" @@ -178,9 +167,6 @@ impl DelegateTool { let task_values: Vec<&Value> = match args.get("tasks").and_then(Value::as_array) { Some(tasks) if !tasks.is_empty() => tasks.iter().collect(), Some(_) => return Ok(failure("tasks must not be empty")), - None if legacy_parallel => { - return Ok(failure("legacy parallel mode requires a tasks array")); - } None => vec![args], }; let mut configs = Vec::with_capacity(task_values.len()); @@ -228,38 +214,15 @@ impl DelegateTool { match mode { ExecutionMode::Foreground => { - let all_named = configs.iter().all(|config| config.target.is_some()); - let results = if all_named && let Some(coordinator) = self.coordinator.as_ref() { - coordinator - .delegate_foreground(context, configs) - .await - .map_err(|error| anyhow::anyhow!(error.to_string()))? - } else { - if configs.iter().any(|config| config.target.is_some()) { - return Ok(failure( - "mixed named and legacy general batches are not supported", - )); - } - let mut results = self - .sub_agent_manager - .run_foreground_batch(configs, context) - .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; - // Legacy general compatibility: tell the model the - // unnamed path is deprecated so it migrates to named - // Agents (which get durable runs, fixed tool sets and - // per-run resource scopes). - for result in results.iter_mut() { - if matches!(result.status, TaskStatus::Completed) { - result.content = format!( - "{}\n\n[提示] 无 target 的通用 Agent 是兼容模式：不持久化、不可审计。\ - 建议为固定角色创建具名 Agent definition 并使用 target 委托。", - result.content - ); - } - } - results + let Some(coordinator) = self.coordinator.as_ref() else { + return Ok(failure( + "delegate requires agent_orchestration to be enabled (named Agents only)", + )); }; + let results = coordinator + .delegate_foreground(context, configs) + .await + .map_err(|error| anyhow::anyhow!(error.to_string()))?; let payload: Vec<_> = results .into_iter() .map(|result| { @@ -297,91 +260,23 @@ impl DelegateTool { "child Agents cannot create background runs in the current implementation", )); } - let mut config = configs.into_iter().next().expect("checked non-empty"); - if config.target.is_some() { - let Some(coordinator) = self.coordinator.as_ref() else { - return Ok(failure( - "named background Agents require agent_orchestration to be enabled", - )); - }; - config.session_id = context - .agent - .as_ref() - .map(|agent| agent.root_session_id.clone()) - .or_else(|| context.session_id.clone()); - return match coordinator.delegate_background(context, config).await { - Ok(run_id) => Ok(success(json!({ - "status": "accepted", - "runs": vec![json!({ "run_id": run_id, "status": "queued" })] - }))), - Err(error) => Ok(failure(error.to_string())), - }; + let Some(coordinator) = self.coordinator.as_ref() else { + return Ok(failure( + "delegate requires agent_orchestration to be enabled (named Agents only)", + )); + }; + let config = configs.into_iter().next().expect("checked non-empty"); + match coordinator.delegate_background(context, config).await { + Ok(run_id) => Ok(success(json!({ + "status": "accepted", + "runs": vec![json!({ "run_id": run_id, "status": "queued" })] + }))), + Err(error) => Ok(failure(error.to_string())), } - let routing = crate::agent::sub_agent::get_delegate_context().map_err(|_| { - anyhow::anyhow!("background delegate requires an active root Agent worker") - })?; - let task_id = self - .sub_agent_manager - .run_background(config, routing) - .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; - Ok(success(json!({ - "status": "accepted", - "runs": vec![json!({ "run_id": task_id, "status": "queued" })] - }))) } } } - async fn handle_check_task(&self, args: &Value) -> anyhow::Result { - let task_id = required_task_id(args)?; - let Some(task) = self.sub_agent_manager.check_task(task_id).await else { - return Ok(failure(format!("task not found: {task_id}"))); - }; - Ok(success(json!({ - "task_id": task.id, - "status": task.status, - "task": task.prompt, - "result": task.result, - "error": task.error, - "started_at": task.started_at, - "finished_at": task.finished_at - }))) - } - - async fn handle_cancel_task(&self, args: &Value) -> anyhow::Result { - let task_id = required_task_id(args)?; - match self.sub_agent_manager.cancel_task(task_id).await { - Ok(true) => Ok(success( - json!({ "task_id": task_id, "status": "cancelled" }), - )), - Ok(false) => Ok(failure(format!( - "cannot cancel task {task_id}; it is terminal or does not exist" - ))), - Err(error) => Ok(failure(format!("cancel failed: {error}"))), - } - } - - async fn handle_list_tasks( - &self, - context: &ToolExecutionContext, - ) -> anyhow::Result { - let session_id = context - .agent - .as_ref() - .map(|agent| agent.root_session_id.as_str()) - .or(context.session_id.as_deref()) - .ok_or_else(|| anyhow::anyhow!("delegate context is not session-bound"))?; - let tasks = self.sub_agent_manager.list_tasks(session_id).await; - Ok(success( - json!({ "tasks": tasks.into_iter().map(|task| json!({ - "task_id": task.id, - "status": task.status, - "task": task.prompt, - "created_at": task.created_at - })).collect::>() }), - )) - } } #[async_trait] @@ -398,11 +293,6 @@ impl Tool for DelegateTool { json!({ "type": "object", "properties": { - "action": { - "type": "string", - "enum": ["run", "check_task", "cancel_task", "list_tasks"], - "description": "Compatibility task-management actions remain available during migration; omit for run" - }, "target": self.task_schema()["properties"]["target"].clone(), "task": { "type": "string", "description": "Single delegated task" }, "context": { "type": "string", "description": "Explicit context for the child Agent" }, @@ -418,19 +308,17 @@ impl Tool for DelegateTool { "description": "Independent tasks; execution is concurrent and results preserve request order" }, "plan_item_id": { "type": "string" }, - "task_id": { "type": "string", "description": "Legacy task-management action target" }, "allowed_tools": { "type": "array", "items": { "type": "string" }, - "description": "Deprecated; only narrows the legacy general Agent and never expands named Agent permissions" + "description": "Only narrows the target Definition's tool set; never expands permissions" }, "max_iterations": { "type": "integer", "minimum": 1 }, "timeout_secs": { "type": "integer", "minimum": 1 } }, "anyOf": [ { "required": ["task"] }, - { "required": ["tasks"] }, - { "required": ["action"] } + { "required": ["tasks"] } ] }) } @@ -454,24 +342,10 @@ impl Tool for DelegateTool { context: &ToolExecutionContext, args: Value, ) -> anyhow::Result { - let action = args.get("action").and_then(Value::as_str).unwrap_or("run"); - let result = match action { - "run" => self.handle_run(&args, context).await?, - "check_task" => self.handle_check_task(&args).await?, - "cancel_task" => self.handle_cancel_task(&args).await?, - "list_tasks" => self.handle_list_tasks(context).await?, - other => failure(format!("unknown delegate action '{other}'")), - }; - Ok(result.into()) + self.handle_run(&args, context).await.map(Into::into) } } -fn required_task_id(args: &Value) -> anyhow::Result<&str> { - args.get("task_id") - .and_then(Value::as_str) - .ok_or_else(|| anyhow::anyhow!("missing required parameter: task_id")) -} - fn status_projection(status: &TaskStatus) -> (&'static str, Option<&str>) { match status { TaskStatus::Completed => ("completed", None), @@ -503,7 +377,6 @@ mod tests { use std::collections::HashMap; fn manager() -> Arc { - let (notify_tx, _) = tokio::sync::mpsc::unbounded_channel(); Arc::new(SubAgentManager::new( crate::config::LLMProviderConfig { provider_type: "openai".to_string(), @@ -524,10 +397,7 @@ mod tests { }, Arc::new(crate::tools::ToolRegistry::new()), None, - notify_tx, - 1, None, - crate::task_supervisor::TaskSupervisor::new(), )) } diff --git a/src/tools/todo.rs b/src/tools/todo.rs index 29d8e14..a22ec5d 100644 --- a/src/tools/todo.rs +++ b/src/tools/todo.rs @@ -15,9 +15,13 @@ impl TodoTool { 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")) + fn session_id(context: &crate::tools::ToolExecutionContext) -> anyhow::Result { + context + .agent + .as_ref() + .map(|agent| agent.root_session_id.clone()) + .or_else(|| context.session_id.clone()) + .ok_or_else(|| anyhow::anyhow!("todo requires a session-bound context")) } } @@ -64,10 +68,20 @@ impl Tool for TodoTool { } async fn execute(&self, args: serde_json::Value) -> anyhow::Result { + self.execute_with_context(&crate::tools::ToolExecutionContext::default(), args) + .await + .map(|output| output.result) + } + + async fn execute_with_context( + &self, + context: &crate::tools::ToolExecutionContext, + args: serde_json::Value, + ) -> anyhow::Result { let action = args["action"] .as_str() .ok_or_else(|| anyhow::anyhow!("missing action"))?; - let ctx = Self::context()?; + let session_id = Self::session_id(context)?; let expected = args["expected_version"].as_i64(); let result = match action { "create" => { @@ -79,28 +93,29 @@ impl Tool for TodoTool { .filter_map(|value| value.as_str().map(str::to_string)) .collect::>(); self.work_manager - .create_plan(&ctx.session_id, objective, &items) + .create_plan(&session_id, objective, &items) .await } - "view" => match self.work_manager.active_plan(&ctx.session_id).await? { + "view" => match self.work_manager.active_plan(&session_id).await? { Some(plan) => Ok(plan), None => { return Ok(ToolResult { success: true, output: "当前 session 没有 active plan".to_string(), error: None, - }); + } + .into()); } }, "append" => { self.work_manager - .append_item(&ctx.session_id, required_str(&args, "title")?, expected) + .append_item(&session_id, required_str(&args, "title")?, expected) .await } "update" => { self.work_manager .update_item( - &ctx.session_id, + &session_id, required_str(&args, "item_id")?, required_str(&args, "status")?, args["summary"].as_str(), @@ -110,7 +125,7 @@ impl Tool for TodoTool { } "close" => { self.work_manager - .close_plan(&ctx.session_id, required_str(&args, "status")?, expected) + .close_plan(&session_id, required_str(&args, "status")?, expected) .await } _ => return Err(anyhow::anyhow!("unknown todo action: {action}")), @@ -121,12 +136,14 @@ impl Tool for TodoTool { success: true, output: render_plan(&plan), error: None, - }), + } + .into()), Err(error) => Ok(ToolResult { success: false, output: String::new(), error: Some(error.to_string()), - }), + } + .into()), } } }