From b13450498b33efcc0f1379b4cf01bd8423f46279 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Thu, 13 Aug 2026 18:08:07 +0800 Subject: [PATCH] refactor: remove sleep tool and wake-aware steering machinery Drop the model-callable sleep tool and its TurnWakeup publisher/handle state. Async background completions and user input already inject through steer-at-safe-boundary or the queued continuation Turn, so the sleep path only misled agents into busy-waiting on non-actionable queue wakes. Cancellation still normalizes running tool blocks to Cancelled. --- AGENTS.md | 4 +- README.md | 7 +- docs/ARCHITECTURE.md | 4 +- resources/agents/general-purpose.md | 1 - .../skills/about-picobot/references/tools.md | 8 +- src/agent/steering.rs | 92 ---- src/session/session.rs | 239 +------- src/session/turn.rs | 6 +- src/tools/mod.rs | 8 +- src/tools/sleep.rs | 520 ------------------ src/tools/traits.rs | 23 - 11 files changed, 38 insertions(+), 874 deletions(-) delete mode 100644 src/tools/sleep.rs diff --git a/AGENTS.md b/AGENTS.md index 23b9f3e..7196988 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, tool/Skill allowlists, and delegation edges before activation. Named Agents support foreground (single/batch) and Root-initiated single background execution with durable run/inbox delivery; a built-in general-purpose definition is released to `~/.picobot/agents/` on first run. Background batches and nested background remain restricted +- **AgentCatalog** is immutable per runtime generation; candidate preparation strictly validates trusted Markdown definitions, Provider profiles, tool/Skill allowlists, and delegation edges before activation. Sub-Agent orchestration is an intrinsic, always-on mechanism (no feature switch). Named Agents support foreground (single/batch) and Root-initiated single background execution with durable run/inbox delivery; a built-in general-purpose definition is released to `~/.picobot/agents/` on first run. Background batches and nested background 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 @@ -107,7 +107,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **Provider reasoning state** is private replay data: persist it, replay it only to the matching provider, and never expose it to clients, channels, or logs - **Tools** are executed by `AgentLoop`; every invocation is normalized to `ToolOutput` and passes through `ToolOutputProcessor`. Plain `ToolResult` implementations use the default conversion, while artifact-producing tools declare model/user audience explicitly; model capability checks, final-reply attachment, channel delivery, and Provider serialization stay outside tools - **Delegated tool access**: a named Agent's tool set is decided solely by its definition file (admin-authored). `delegate`, `emit_signal`, `get_skill` and `agent_task` are runtime-injected and must never be declared in `tools` (`get_skill` is the scoped-skill switch); `allowed_tools` can only narrow the definition, never expand it -- **Sleep tool** is a cancellable foreground wait bounded to 24 hours; longer or durable delays belong to Scheduler/background work, and cancelling a Turn must normalize active tool blocks to `Cancelled` +- **No foreground wait tool**: Agents wait for asynchronous work by ending the Turn and letting queued completions/signals open a continuation Turn, or by polling status tools; there is no model-callable `sleep`/wait tool. Cancelling a Turn must still normalize active tool blocks to `Cancelled` - **Stateful tools** receive `ToolExecutionContext`; browser calls without `persistent_id` map each PicoBot dialog to an opaque transient agent-browser session. For long-running work an Agent may autonomously create a persistent identity and must pass its validated ID on every related action; the same ID shares one agent-browser session and serialization gate across dialogs, while different IDs have independent sessions/gates. `browser_profiles` may create IDs, persist bounded semantic labels, list, or delete only validated IDs beneath the configured profile root. Do not introduce a global persistence mode switch, a default persistent ID, automatic per-dialog persistent Profiles, Fantoccini, ChromeDriver, WebDriver, model-controlled raw agent-browser session IDs, or arbitrary Profile paths - **Health diagnostics** are read-only and share `HealthService` across `picobot health`, the `health` tool, and `/health`; checks must not install/fix dependencies, call Provider APIs, or expose secrets diff --git a/README.md b/README.md index a8b1aff..ca4ce1e 100644 --- a/README.md +++ b/README.md @@ -335,7 +335,6 @@ PicoBot 有两类记忆： | 工具 | 说明 | |------|------| | `calculator` | 数学表达式和统计计算 | -| `sleep` | 暂停当前 Agent 工具调用 0～86400 秒；可由用户停止，不用于持久调度 | | `file_read` / `file_write` / `file_edit` | 文件读写和编辑；`file_read` 读取受支持图片时可将图片直接提供给多模态模型 | | `file_search` / `content_search` | 文件名和内容搜索 | | `bash` | 在 workspace 中执行 Shell 命令 | @@ -375,7 +374,7 @@ Skill 是包含 `SKILL.md` 的目录。加载优先级从高到低： | `providers` | LLM Provider 配置 | | `models` | 模型参数与输入能力 | | `agents` | Agent 使用哪个 provider/model | -| `agent_orchestration` | 具名子 Agent 定义目录、Root 委托白名单和编排上限；默认关闭 | +| `agent_orchestration` | 具名子 Agent 定义目录与编排上限 | | `gateway` | HTTP/WebSocket、数据库、调度器、后台任务限制 | | `client` | CLI 客户端默认 Gateway URL | | `channels` | 渠道配置，目前主要是飞书/Lark | @@ -408,7 +407,7 @@ Skill 是包含 `SKILL.md` 的目录。加载优先级从高到低： ### 具名子 Agent（Phase 1） -启用 `agent_orchestration.enabled` 后，PicoBot 会在 Gateway 候选运行代构造时从配置目录下的 `definitions_dir` 加载 `*.md`。相对路径按 `config.json` 所在目录解析，且 canonical path 不得逃逸该目录；角色文件、Provider profile、工具、Skill 和委托边任一无效都会拒绝启动或热重载。支持具名 `foreground`（单/批量）与 Root 发起的具名 `background`（单任务或 `tasks[]` 批量）：每个 run 独立落库、预留 completion 槽、完成后由主 Agent 的 continuation Turn 单独汇总（空闲时完成即返回），可配合 `emit_signal`（queue/steer）推送内部信号。子 Agent 发起的 background 尚未开放；未启用编排时无法委托（旧匿名 general 已移除）。 +PicoBot 在 Gateway 候选运行代构造时从配置目录下的 `definitions_dir` 加载 `*.md`（子 Agent 编排是内在机制，始终启用）。相对路径按 `config.json` 所在目录解析，且 canonical path 不得逃逸该目录；角色文件、Provider profile、工具、Skill 和委托边任一无效都会拒绝启动或热重载。支持具名 `foreground`（单/批量）与 Root 发起的具名 `background`（单任务或 `tasks[]` 批量）：每个 run 独立落库、预留 completion 槽、完成后由主 Agent 的 continuation Turn 单独汇总（空闲时完成即返回），可配合 `emit_signal`（queue/steer）推送内部信号。子 Agent 发起的 background 尚未开放（旧匿名 general 已移除）。 ```md --- @@ -430,7 +429,7 @@ limits: 你是一名严谨的研究 Agent，只返回与任务有关的结论和证据。 ``` -每个具名 Agent 的工具集完全由其 Markdown `tools` 列表决定（管理员显式授权），不再有工具侧的可派发门槛；也可内联 `provider`/`model` 直接指定模型（或沿用 `llm_profile` 引用顶层 `agents` key）。`delegate`/`emit_signal`/`get_skill`/`agent_task` 为运行时注入工具，不能写进 `tools`（分别由 `delegates`/`signal`/`skills` 字段派生），`get_skill` 例外作为启用 scoped skill 的开关。WebUI「子 Agent」页可直接增删改定义、启停并选择工具/Skill/Provider/Model。 +每个具名 Agent 的工具集完全由其 Markdown `tools` 列表决定（管理员显式授权），不再有工具侧的可派发门槛；也可内联 `provider`/`model` 直接指定模型（或沿用 `llm_profile` 引用顶层 `agents` key）。`delegate`/`emit_signal`/`get_skill`/`agent_task` 为运行时注入工具，不能写进 `tools`（分别由 `delegates`/`signal`/`skills` 字段派生），`get_skill` 例外作为启用 scoped skill 的开关。每个定义可用 `enabled: false` 单独禁用（保留在磁盘但不加载）。主 Agent 可委托给任意具名子 Agent；子 Agent 能否继续委托由 `delegates` 决定——不写该字段时默认仅可委托内置 `general-purpose`，写 `[]` 表示不可继续委托，写 `["*"]` 表示可委托任意子代理，写列表则按列表指定（self 与祖先在运行时始终被拒绝）。WebUI「子 Agent」页可直接增删改定义、启停并选择工具/Skill/Provider/Model 与委托范围。 更完整的配置字段说明见 [resources/skills/about-picobot/references/config.md](resources/skills/about-picobot/references/config.md)。 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3f91aee..2adf2c9 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -2,7 +2,7 @@ 本文档描述 PicoBot 当前实现的运行时边界、数据流、并发模型和演进约束。它面向维护者和后续参与改进的 Agent，是代码架构的主入口；行为细节仍以代码和测试为最终依据。 -流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递的详细设计与取舍见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。用户输入路由、Session 执行拆分、终态投递确认和历史增量校准的重构方案见 [MESSAGE_FLOW_REFACTOR_DESIGN.md](MESSAGE_FLOW_REFACTOR_DESIGN.md)。配置运行代、重载边界和失败语义见 [CONFIG_HOT_RELOAD_DESIGN.md](CONFIG_HOT_RELOAD_DESIGN.md)。具名子 Agent、委托图、后台收件箱、`queue`/`steer` 信号和可唤醒 `sleep` 的升级提案见 [SUB_AGENT_ORCHESTRATION_DESIGN.md](SUB_AGENT_ORCHESTRATION_DESIGN.md)。 +流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递的详细设计与取舍见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。用户输入路由、Session 执行拆分、终态投递确认和历史增量校准的重构方案见 [MESSAGE_FLOW_REFACTOR_DESIGN.md](MESSAGE_FLOW_REFACTOR_DESIGN.md)。配置运行代、重载边界和失败语义见 [CONFIG_HOT_RELOAD_DESIGN.md](CONFIG_HOT_RELOAD_DESIGN.md)。具名子 Agent、委托图、后台收件箱、结果传递机制与 `queue`/`steer` 信号的设计见 [SUB_AGENT_DESIGN.md](SUB_AGENT_DESIGN.md)。 ## 1. 设计目标 @@ -321,7 +321,7 @@ WebUI/TUI 文件字节通过受鉴权的 HTTP 接口流式传输，WebSocket 只 5. 长操作应有超时；后台执行应交给 SubAgentManager/TaskSupervisor。 6. 需要跨调用保存外部状态时，实现 `execute_with_context` 并按 session 隔离；不得让模型控制底层全局 session ID。 -内置 `sleep` 只暂停当前前台工具 Future，允许 0～86400 秒且不持久化；`/stop`、Scheduler/SubAgent 超时和 Supervisor shutdown 通过丢弃外层 Future 取消计时。Turn 进入 `Cancelled` 时必须把仍为 `Running` 的工具块同步归约为 `Cancelled`，避免终态快照继续显示工具执行中。超过 24 小时或需要跨重启的等待必须使用 Scheduler/后台任务。 +没有模型可调用的前台 `sleep`/等待工具：Agent 等待异步工作时，应结束当前 Turn 让排队完成/信号开启续接 Turn，或轮询状态工具。Turn 进入 `Cancelled` 时仍必须把 `Running` 的工具块同步归约为 `Cancelled`。需要跨重启的可靠延迟必须使用 Scheduler/后台任务。 ### 新增 Provider diff --git a/resources/agents/general-purpose.md b/resources/agents/general-purpose.md index 45d2938..e4a8902 100644 --- a/resources/agents/general-purpose.md +++ b/resources/agents/general-purpose.md @@ -9,7 +9,6 @@ tools: - content_search - web_fetch - calculator - - sleep --- # General Purpose Agent diff --git a/resources/skills/about-picobot/references/tools.md b/resources/skills/about-picobot/references/tools.md index 74ef2c9..dba4b70 100644 --- a/resources/skills/about-picobot/references/tools.md +++ b/resources/skills/about-picobot/references/tools.md @@ -137,7 +137,7 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 | 参数 | 必填 | 说明 | |------|------|------| -| `target` | 具名 Agent 必填 | `root_delegates` 或当前 Agent Definition 允许的目标 ID | +| `target` | 具名 Agent 必填 | 目标 Agent ID；主 Agent 可委托给任意具名子 Agent，子 Agent 按自身 Definition 的 `delegates` 白名单决定 | | `task` | 单任务必填 | 明确、独立、可验收的子任务 | | `context` | 否 | 子 Agent 所需的显式事实；不会继承完整主会话历史 | | `mode` | 否 | `foreground`（默认）或 `background`；批量并发不是第三种 mode | @@ -211,12 +211,6 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 用于交互式程序和需要保持状态的长运行命令。`action` 支持 `spawn`、`write`、`read`、`kill`、`list`；`write/read/kill` 需要 `session_id`。Gateway 进程退出时 PTY manager 会清理子进程。 -## sleep — 前台等待 - -参数 `seconds` 接受 0～86400 的整数。工具只暂停当前 Agent 工具调用，不持久化、不发送消息，也不保证跨进程重启继续；用户 `/stop`、Scheduler/SubAgent 超时和 Gateway shutdown 都会取消等待。超过 24 小时或需要可靠延迟执行时应使用 Scheduler。 - -主 Agent（root interactive Turn）的 sleep 是 wake-aware：当前 session 收到任何新输入（用户 steer/queue、后台 Agent 的 steer 信号或排队结果）都会提前结束等待。Steer 唤醒会告知来源 run/agent 与安全摘要，并在当前 Turn 的下一个安全边界注入；queue 唤醒只说明类型与数量，内容不会进入当前 Turn。子 Agent run 与 continuation Turn 没有 session 输入通道，其 sleep 只响应 timer/cancel。 - ## http_request / web_fetch — HTTP 和 Web 工具 `http_request` 支持 GET/POST/PUT/DELETE/PATCH、headers 和字符串 body；`web_fetch` 提取 HTML/JSON 的可读文本。两者校验 URL 与 DNS 解析结果，阻止回环、私网、link-local 和本地域名，并禁用自动重定向，以降低 SSRF 风险。 diff --git a/src/agent/steering.rs b/src/agent/steering.rs index 1ef3798..3738943 100644 --- a/src/agent/steering.rs +++ b/src/agent/steering.rs @@ -46,98 +46,6 @@ impl TurnInputSource { } } -impl From<&TurnInputSource> for WakeupSource { - fn from(source: &TurnInputSource) -> Self { - match source { - TurnInputSource::User => WakeupSource::UserSteer, - TurnInputSource::AgentSignal { run_id, agent_id } => WakeupSource::AgentSignal { - run_id: run_id.clone(), - agent_id: agent_id.clone(), - }, - TurnInputSource::AgentCompletion { run_id, agent_id } => { - WakeupSource::AgentCompletion { - run_id: run_id.clone(), - agent_id: agent_id.clone(), - } - } - } - } -} - -/// What woke a root-interactive sleep. Queue wakes carry no content: the -/// model only learns a type/count, never the payload. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum WakeupSource { - UserSteer, - UserQueue, - AgentSignal { run_id: String, agent_id: String }, - AgentCompletion { run_id: String, agent_id: String }, - AgentQueue, -} - -/// Snapshot published to sleeping root Turns whenever a new input is -/// durably admitted anywhere on the session's receive surface. -#[derive(Debug, Clone, Default)] -pub struct TurnWakeupState { - pub revision: u64, - pub pending_user_steer: usize, - pub pending_user_queue: usize, - pub pending_agent_steer: usize, - pub pending_agent_queue: usize, - pub latest_source: Option, - /// Safe, model-visible preview for steer wakes only. Queue wakes never - /// carry content. - pub latest_safe_preview: Option, -} - -impl TurnWakeupState { - pub fn pending_total(&self) -> usize { - self.pending_user_steer - .saturating_add(self.pending_user_queue) - .saturating_add(self.pending_agent_steer) - .saturating_add(self.pending_agent_queue) - } -} - -/// Root-Turn-side receiver used by wake-aware tools (sleep). -#[derive(Debug, Clone)] -pub struct TurnWakeupHandle { - pub receiver: tokio::sync::watch::Receiver, -} - -/// Session-side publisher for the active Turn. Admission points bump the -/// revision and `send_replace` AFTER the durable fact is visible, so a -/// waking sleep can always observe the input it was told about. -#[derive(Debug, Clone)] -pub struct TurnWakeupPublisher { - sender: tokio::sync::watch::Sender, -} - -impl TurnWakeupPublisher { - pub fn new() -> Self { - let (sender, _) = tokio::sync::watch::channel(TurnWakeupState::default()); - Self { sender } - } - - pub fn subscribe(&self) -> TurnWakeupHandle { - TurnWakeupHandle { - receiver: self.sender.subscribe(), - } - } - - pub fn publish(&self, state: TurnWakeupState) { - let mut state = state; - state.revision = state.revision.saturating_add(1); - let _ = self.sender.send_replace(state); - } -} - -impl Default for TurnWakeupPublisher { - fn default() -> Self { - Self::new() - } -} - /// How the input reached the mailbox. Queue inputs belong to the next Turn; /// only Steer entries are drained by the active Turn. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/src/session/session.rs b/src/session/session.rs index 0e72e66..7675d54 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -663,8 +663,6 @@ struct ActiveTurnEmitter { /// `AgentTurnContext`; keeping it on the session handle makes admission /// atomic with `/stop` and worker cleanup. steering: Arc, - /// Watch publisher for wake-aware tools (sleep) of this root Turn. - wakeup: crate::agent::steering::TurnWakeupPublisher, /// Original inbound tasks for accepted steering messages. ChatMessage /// intentionally carries only durable history fields, so this side map /// preserves channel context and rich MediaItem metadata if a terminal @@ -692,11 +690,7 @@ struct AgentTask { fn steer_input_from_event( event: &crate::storage::agent_inbox::AgentInboxEventRecord, now: i64, -) -> ( - TurnInput, - crate::agent::steering::WakeupSource, - Option, -) { +) -> TurnInput { use crate::agent::steering::InputDelivery; use crate::storage::agent_inbox::AgentEventType; let payload: serde_json::Value = @@ -707,7 +701,7 @@ fn steer_input_from_event( .and_then(serde_json::Value::as_str) .unwrap_or("unknown") .to_string(); - let (source, wakeup_preview, content) = match event.event_type { + let (source, content) = match event.event_type { AgentEventType::Signal => { let severity = event.severity.clone().unwrap_or_else(|| "info".to_string()); let summary = payload @@ -729,7 +723,6 @@ fn steer_input_from_event( run_id: run_id.clone(), agent_id: agent_id.clone(), }, - (!summary.is_empty()).then_some(summary), content, ) } @@ -748,12 +741,11 @@ fn steer_input_from_event( run_id: run_id.clone(), agent_id: agent_id.clone(), }, - None, content, ) } }; - let input = TurnInput { + TurnInput { id: format!("steer:{}", event.id), sequence: 0, source, @@ -764,9 +756,7 @@ fn steer_input_from_event( received_at: now, message_source: None, lease_token: Some(event.lease_token.clone().unwrap_or_default()), - }; - let wakeup_source = crate::agent::steering::WakeupSource::from(&input.source); - (input, wakeup_source, wakeup_preview) + } } /// Move terminally pending steering into the worker's local FIFO. The @@ -1297,7 +1287,6 @@ impl Session { turn_id: turn_id.to_string(), emitter, steering: TurnMailbox::new_shared(), - wakeup: crate::agent::steering::TurnWakeupPublisher::new(), recovery: StdArc::new(StdMutex::new(HashMap::new())), }); } @@ -1968,11 +1957,7 @@ impl SessionManager { ) .map_err(|error| AgentError::Other(format!("failed to load Agent catalog: {error}")))?, ); - let execution_gate = if catalog_preparation.config.enabled { - crate::agent::gate::ExecutionGate::new(&catalog_preparation.config) - } else { - crate::agent::gate::ExecutionGate::unbounded() - }; + let execution_gate = crate::agent::gate::ExecutionGate::new(&catalog_preparation.config); // Create SubAgentManager and register DelegateTool let sub_agent_manager = Arc::new( @@ -1989,26 +1974,22 @@ impl SessionManager { let mut delegate_tool = crate::tools::DelegateTool::new(sub_agent_manager.clone()); let inbox_notifier = crate::agent::AgentInboxNotifier::new(); let agent_projection_hub = Arc::new(crate::agent::AgentProjectionHub::new()); - let agent_coordinator = if agent_catalog.enabled() { - let coordinator = crate::agent::AgentCoordinator::new( - storage.clone(), - sub_agent_manager.clone(), - work_manager.clone(), - inbox_notifier.clone(), - agent_projection_hub.clone(), - execution_gate.clone(), - admission.clone(), - task_supervisor.clone(), - catalog_preparation.runtime_generation, - &catalog_preparation.config, - ); - tools.register(crate::tools::AgentTaskTool::new(coordinator.clone())); - delegate_tool = delegate_tool.with_coordinator(coordinator.clone()); - sub_agent_manager.bind_coordinator(&coordinator); - Some(coordinator) - } else { - None - }; + let coordinator = crate::agent::AgentCoordinator::new( + storage.clone(), + sub_agent_manager.clone(), + work_manager.clone(), + inbox_notifier.clone(), + agent_projection_hub.clone(), + execution_gate.clone(), + admission.clone(), + task_supervisor.clone(), + catalog_preparation.runtime_generation, + &catalog_preparation.config, + ); + tools.register(crate::tools::AgentTaskTool::new(coordinator.clone())); + delegate_tool = delegate_tool.with_coordinator(coordinator.clone()); + sub_agent_manager.bind_coordinator(&coordinator); + let agent_coordinator = Some(coordinator); tools.register(delegate_tool); tools.register(crate::tools::ReloadConfigTool::new(reload.clone())); @@ -3296,21 +3277,6 @@ impl SessionManager { ); match active.steering.try_push_user(input) { Ok(()) => { - // Wake-aware sleep: the input is durably visible - // in the mailbox before the publish. - if let Some(active) = guard.active_turn_emitter.as_ref() { - active - .wakeup - .publish(crate::agent::steering::TurnWakeupState { - pending_user_steer: active.steering.user_pending_count(), - pending_agent_steer: active.steering.agent_pending_count(), - latest_source: Some( - crate::agent::steering::WakeupSource::UserSteer, - ), - latest_safe_preview: None, - ..Default::default() - }); - } return Ok(HandleResult::AgentProcessing); } Err(_) => { @@ -3411,24 +3377,6 @@ impl SessionManager { AgentError::Other("agent worker spawn+send failed irrecoverably".to_string()) })?; } - // Wake-aware sleep: a queued user input must wake a sleeping root - // Turn (the content stays in the queue for the next Turn). - if let Some(active) = guard.active_turn_emitter.as_ref() { - let queued = guard - .agent_tx - .as_ref() - .map(|tx| tx.max_capacity() - tx.capacity()) - .unwrap_or(1) - .max(1); - active - .wakeup - .publish(crate::agent::steering::TurnWakeupState { - pending_user_queue: queued, - latest_source: Some(crate::agent::steering::WakeupSource::UserQueue), - latest_safe_preview: None, - ..Default::default() - }); - } Ok(HandleResult::AgentProcessing) } } @@ -3780,12 +3728,10 @@ fn spawn_agent_worker( let initial_turn = turn_controller.snapshot(); let steering = TurnMailbox::new_shared(); let recovery = StdArc::new(StdMutex::new(HashMap::new())); - let turn_wakeup = crate::agent::steering::TurnWakeupPublisher::new(); guard.active_turn_emitter = Some(ActiveTurnEmitter { turn_id: initial_turn.id.0.clone(), emitter: turn_emitter.clone(), steering: steering.clone(), - wakeup: turn_wakeup, recovery: recovery.clone(), }); @@ -3984,24 +3930,14 @@ fn spawn_agent_worker( let pending_turn_deliveries = Arc::new(std::sync::Mutex::new(Vec::new())); let scoped_turn_deliveries = pending_turn_deliveries.clone(); let steering_for_process = steering.clone(); - let wakeup_handle = { - let guard = session.lock().await; - guard - .active_turn_emitter - .as_ref() - .map(|active| active.wakeup.subscribe()) - }; let process_gate = execution_gate.clone(); let turn_token_for_process = turn_token.clone(); let process_future = async move { let response_session_id = unified_str2.clone(); - let mut tool_context = ToolExecutionContext::for_session(&response_session_id) + let tool_context = ToolExecutionContext::for_session(&response_session_id) .with_turn_id(agent_turn.turn_id.clone()) .with_cancellation(turn_token_for_process) .with_execution_gate(process_gate.clone()); - if let Some(handle) = wakeup_handle { - tool_context = tool_context.with_turn_wakeup(handle); - } let process_result = agent .process_streaming_with_context( history_out.clone(), @@ -4725,25 +4661,8 @@ impl crate::agent::AgentInboxWakeTarget for SessionManager { // admission; everything that cannot be admitted stays pending // for the queue lane. if has_active_turn { - let outcome = self - .try_steer_inbox_events(&session, session_id, revision) + self.try_steer_inbox_events(&session, session_id, revision) .await; - // Events remain pending for the queue lane: wake-aware - // sleep must end even though nothing entered the Turn. - if !matches!(outcome, SteerAdmission::Activated) { - let guard = session.lock().await; - if let Some(active) = guard.active_turn_emitter.as_ref() { - active - .wakeup - .publish(crate::agent::steering::TurnWakeupState { - pending_agent_queue: 1, - latest_source: Some( - crate::agent::steering::WakeupSource::AgentQueue, - ), - ..Default::default() - }); - } - } } } let mut guard = session.lock().await; @@ -4823,13 +4742,11 @@ impl SessionManager { }; let mut rejected = Vec::new(); let mut reserved = Vec::new(); - let mut wakeup_sources = Vec::new(); for event in &lease.events { - let (input, wakeup_source, preview) = steer_input_from_event(event, now); + let input = steer_input_from_event(event, now); match mailbox.try_reserve_steer(input, token.clone()) { Ok(()) => { reserved.push(event.id.clone()); - wakeup_sources.push((wakeup_source, preview)); } Err(_) => rejected.push((event.id.clone(), token.clone())), } @@ -4862,24 +4779,6 @@ impl SessionManager { }; if still_active { mailbox.activate_reserved(); - // Wake-aware sleep: publish AFTER the durable admit, so a - // waking tool can observe the input it was told about. - let guard = session.lock().await; - if let Some(active) = guard.active_turn_emitter.as_ref() { - let (latest_source, latest_safe_preview) = wakeup_sources - .into_iter() - .next() - .unwrap_or((crate::agent::steering::WakeupSource::AgentQueue, None)); - active - .wakeup - .publish(crate::agent::steering::TurnWakeupState { - pending_user_steer: active.steering.user_pending_count(), - pending_agent_steer: active.steering.agent_pending_count(), - latest_source: Some(latest_source), - latest_safe_preview, - ..Default::default() - }); - } return SteerAdmission::Activated; } rejected.extend(admitted); @@ -5033,7 +4932,7 @@ impl SessionManager { #[cfg(test)] mod slash_command_tests { use super::{ - AgentTask, SLASH_COMMANDS, Session, pop_lowest_sequence, prepend_pending_steering, + AgentTask, SLASH_COMMANDS, pop_lowest_sequence, prepend_pending_steering, resolve_slash_command, }; use crate::agent::steering::{SteeringPushError, TurnInput, TurnMailbox}; @@ -5219,92 +5118,4 @@ mod slash_command_tests { assert!(mailbox.is_closed()); assert!(mailbox.take_pending().is_empty()); } - - #[tokio::test] - async fn active_turn_wakeup_publisher_reaches_sleep_handles() { - use crate::agent::steering::{TurnWakeupState, WakeupSource}; - use crate::config::LLMProviderConfig; - use crate::memory::MemoryManager; - use crate::session::UnifiedSessionId; - use crate::tools::ToolRegistry; - use std::collections::HashMap; - use std::path::PathBuf; - - let dir = tempfile::tempdir().unwrap(); - let storage = Arc::new( - crate::storage::Storage::new(&dir.path().join("wakeup.db")) - .await - .unwrap(), - ); - let memory_manager = Arc::new(MemoryManager::new( - storage, - "test".to_string(), - "test".to_string(), - )); - let config = LLMProviderConfig { - provider_type: "openai".to_string(), - name: "test".to_string(), - base_url: "http://127.0.0.1".to_string(), - api_key: "test".to_string(), - extra_headers: HashMap::new(), - model_id: "test".to_string(), - temperature: None, - max_tokens: None, - model_extra: HashMap::new(), - max_tool_iterations: 1, - token_limit: 8_192, - workspace_dir: PathBuf::from("."), - input_types: vec!["text".to_string()], - price_input_per_million: None, - price_output_per_million: None, - }; - let session = Arc::new(tokio::sync::Mutex::new( - Session::new( - UnifiedSessionId::new("cli_chat", "chat", "dialog"), - config, - Arc::new(ToolRegistry::new()), - None, - String::new(), - "test".to_string(), - memory_manager, - ) - .await - .unwrap(), - )); - session.lock().await.set_active_turn_for_test("turn-1"); - - // The emitter's publisher is the same one a sleep handle subscribes - // to via `with_turn_wakeup`. - let handle = { - let guard = session.lock().await; - guard - .active_turn_emitter - .as_ref() - .expect("active turn installed") - .wakeup - .subscribe() - }; - let mut rx = handle.receiver.clone(); - assert_eq!(rx.borrow_and_update().pending_total(), 0); - - // User steer publish (what handle_message performs). - { - let guard = session.lock().await; - guard - .active_turn_emitter - .as_ref() - .unwrap() - .wakeup - .publish(TurnWakeupState { - pending_user_steer: 1, - latest_source: Some(WakeupSource::UserSteer), - ..Default::default() - }); - } - assert!(rx.changed().await.is_ok()); - let state = rx.borrow_and_update(); - assert_eq!(state.pending_total(), 1); - assert_eq!(state.latest_source, Some(WakeupSource::UserSteer)); - assert!(state.revision > 0); - } } diff --git a/src/session/turn.rs b/src/session/turn.rs index 6f46dd9..dd3607e 100644 --- a/src/session/turn.rs +++ b/src/session/turn.rs @@ -600,8 +600,8 @@ mod tests { .emit(TurnEvent::ToolStarted { iteration: 0, call: ToolCall { - id: "sleep-call".into(), - name: "sleep".into(), + id: "long-call".into(), + name: "long_tool".into(), arguments: serde_json::json!({"seconds": 60}), }, }) @@ -617,7 +617,7 @@ mod tests { id, status: ToolStatus::Cancelled, .. - } if id == "sleep-call" + } if id == "long-call" )); } diff --git a/src/tools/mod.rs b/src/tools/mod.rs index a5851d7..1d9ee84 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -23,7 +23,6 @@ pub mod registry; pub mod reload_config; pub mod schema; pub mod send_message; -pub mod sleep; pub mod todo; pub mod traits; pub mod web_fetch; @@ -49,12 +48,10 @@ pub use pty::{PtyManager, PtyTool}; pub use registry::ToolRegistry; pub use reload_config::ReloadConfigTool; pub use send_message::SendMessageTool; -pub use sleep::SleepTool; pub use todo::TodoTool; pub use traits::{ - InputInterruptPolicy, OutboundDelivery, OutboundMessenger, ProcessedToolOutput, Tool, - ToolArtifact, ToolArtifactAudience, ToolExecutionContext, ToolOutput, ToolOutputProcessor, - ToolResult, + OutboundDelivery, OutboundMessenger, ProcessedToolOutput, Tool, ToolArtifact, + ToolArtifactAudience, ToolExecutionContext, ToolOutput, ToolOutputProcessor, ToolResult, }; pub use web_fetch::WebFetchTool; @@ -80,7 +77,6 @@ pub fn create_default_tools( ) -> anyhow::Result { let registry = ToolRegistry::new(); registry.register(CalculatorTool::new()); - registry.register(SleepTool::new()); registry.register(FileReadTool::new()); registry.register(FileWriteTool::new()); registry.register(FileEditTool::new()); diff --git a/src/tools/sleep.rs b/src/tools/sleep.rs deleted file mode 100644 index 70b559c..0000000 --- a/src/tools/sleep.rs +++ /dev/null @@ -1,520 +0,0 @@ -use super::traits::{Tool, ToolResult}; -use async_trait::async_trait; -use serde_json::json; -use std::time::Duration; - -use crate::agent::steering::{TurnWakeupState, WakeupSource}; - -const MAX_SLEEP_SECONDS: u64 = 86_400; - -pub struct SleepTool; - -impl SleepTool { - pub fn new() -> Self { - Self - } -} - -impl Default for SleepTool { - fn default() -> Self { - Self::new() - } -} - -fn parse_seconds(args: &serde_json::Value) -> Result { - let seconds = args - .get("seconds") - .and_then(serde_json::Value::as_u64) - .ok_or_else(|| "seconds must be a non-negative integer".to_string())?; - if seconds > MAX_SLEEP_SECONDS { - return Err(format!( - "seconds must not exceed {MAX_SLEEP_SECONDS} (24 hours)" - )); - } - Ok(seconds) -} - -#[async_trait] -impl Tool for SleepTool { - fn input_interrupt_policy(&self) -> crate::tools::InputInterruptPolicy { - crate::tools::InputInterruptPolicy::WakeOnly - } - - fn name(&self) -> &str { - "sleep" - } - - fn description(&self) -> &str { - "Pause the current agent execution for a specified number of whole seconds." - } - - fn parameters_schema(&self) -> serde_json::Value { - json!({ - "type": "object", - "properties": { - "seconds": { - "type": "integer", - "minimum": 0, - "maximum": MAX_SLEEP_SECONDS, - "description": "Number of whole seconds to wait, up to 24 hours." - } - }, - "required": ["seconds"] - }) - } - - 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 seconds = match parse_seconds(&args) { - Ok(seconds) => seconds, - Err(error) => { - return Ok(ToolResult { - success: false, - output: String::new(), - error: Some(error), - } - .into()); - } - }; - - let started = std::time::Instant::now(); - let mut wakeup_rx = context - .turn_wakeup - .as_ref() - .map(|handle| handle.receiver.clone()); - - // Root interactive Turn: if inputs are already pending, do not wait - // at all. The watch revision is monotonic, so an input arriving - // between this check and the select below still fires `changed()`. - if let Some(rx) = wakeup_rx.as_mut() { - let state = rx.borrow_and_update(); - if state.pending_total() > 0 { - return Ok(ToolResult { - success: true, - output: wake_message(&state, started.elapsed(), 0), - error: None, - } - .into()); - } - } - - let outcome = match wakeup_rx.as_mut() { - Some(rx) => { - tokio::select! { - biased; - _ = context.cancellation.cancelled() => { - anyhow::bail!("sleep cancelled"); - } - _ = tokio::time::sleep(Duration::from_secs(seconds)) => { - WakeOutcome::Elapsed - } - changed = rx.changed() => { - let _ = changed; - let state = rx.borrow_and_update(); - WakeOutcome::InputArrived(state.clone()) - } - } - } - // Child runs and continuation Turns have no session input lane: - // their sleep answers only the timer, run cancellation, timeout - // and shutdown. - None => { - tokio::select! { - biased; - _ = context.cancellation.cancelled() => { - anyhow::bail!("sleep cancelled"); - } - _ = tokio::time::sleep(Duration::from_secs(seconds)) => { - WakeOutcome::Elapsed - } - } - } - }; - - let output = match outcome { - WakeOutcome::Elapsed => format!("Slept for {seconds} second(s)."), - WakeOutcome::InputArrived(state) => wake_message(&state, started.elapsed(), seconds), - }; - Ok(ToolResult { - success: true, - output, - error: None, - } - .into()) - } -} - -enum WakeOutcome { - Elapsed, - InputArrived(TurnWakeupState), -} - -/// Build the model-visible wake message. Steer wakes describe the source, -/// run identity and a safe preview; queue wakes only state the type/count and -/// explicitly promise the content stays out of the current Turn. -fn wake_message(state: &TurnWakeupState, waited: std::time::Duration, planned: u64) -> String { - let waited_secs = waited.as_secs(); - let mut message = format!("Sleep 提前结束：已等待 {waited_secs} 秒"); - if planned > 0 { - message.push_str(&format!("（原计划 {planned} 秒）")); - } - message.push('。'); - match &state.latest_source { - Some(WakeupSource::UserSteer) => { - message.push_str(" 收到一条新的用户输入，将在当前 Turn 的下一个安全边界注入。"); - } - Some(WakeupSource::UserQueue) => { - message.push_str(&format!( - " 收到 {} 条排队输入。内容不会进入当前 Turn，将在当前工作结束后的下一 Turn处理。", - state.pending_user_queue.max(1) - )); - } - Some(WakeupSource::AgentSignal { run_id, agent_id }) => { - message.push_str(&format!( - " 收到一条 steer AgentSignal（run_id={run_id}, agent={agent_id}）" - )); - if let Some(preview) = state.latest_safe_preview.as_deref() { - message.push_str(&format!("：{preview}")); - } - message.push_str("。该信号将在当前 Turn 的下一个安全边界注入。"); - } - Some(WakeupSource::AgentCompletion { run_id, agent_id }) => { - message.push_str(&format!( - " 收到一条 steer AgentCompletion（run_id={run_id}, agent={agent_id}），将在当前 Turn 的下一个安全边界注入。" - )); - } - Some(WakeupSource::AgentQueue) | None => { - message.push_str(&format!( - " 收到 {} 条排队输入。内容不会进入当前 Turn，将在当前工作结束后的下一 Turn处理。", - state.pending_agent_queue.max(1) - )); - } - } - message -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::agent::TurnEvent; - use crate::agent::steering::TurnWakeupPublisher; - use crate::providers::ToolCall; - use crate::session::{ToolStatus, TurnBlock, TurnController, TurnStatus}; - use crate::tools::Tool; - use serde_json::json; - use std::time::Duration; - - #[test] - fn exposes_sleep_metadata_and_schema() { - let tool = SleepTool::new(); - let schema = tool.parameters_schema(); - - assert_eq!(tool.name(), "sleep"); - assert!(tool.description().contains("current agent execution")); - assert!(tool.description().contains("whole seconds")); - assert_eq!(schema["type"], "object"); - assert_eq!(schema["required"], json!(["seconds"])); - assert_eq!(schema["properties"]["seconds"]["type"], "integer"); - assert_eq!(schema["properties"]["seconds"]["minimum"], 0); - assert_eq!( - schema["properties"]["seconds"]["maximum"], - MAX_SLEEP_SECONDS - ); - assert!(schema.get("additionalProperties").is_none()); - assert!(!tool.read_only()); - assert!(!tool.concurrency_safe()); - assert!(!tool.exclusive()); - assert_eq!( - tool.input_interrupt_policy(), - crate::tools::InputInterruptPolicy::WakeOnly - ); - } - - #[tokio::test] - async fn zero_seconds_returns_exact_success() { - let result = SleepTool::new() - .execute(json!({"seconds": 0})) - .await - .unwrap(); - - assert!(result.success); - assert_eq!(result.output, "Slept for 0 second(s)."); - assert_eq!(result.error, None); - } - - #[tokio::test] - async fn rejects_invalid_seconds() { - let invalid_args = [ - json!({}), - json!({"seconds": -1}), - json!({"seconds": 0.5}), - json!({"seconds": "1"}), - json!({"seconds": 18_446_744_073_709_552_000.0_f64}), - json!({"seconds": MAX_SLEEP_SECONDS + 1}), - ]; - - for args in invalid_args { - let result = SleepTool::new().execute(args).await.unwrap(); - assert!(!result.success); - assert!(result.output.is_empty()); - assert!(result.error.is_some()); - } - } - - #[test] - fn accepts_24_hour_boundary() { - assert_eq!( - parse_seconds(&json!({"seconds": MAX_SLEEP_SECONDS})), - Ok(MAX_SLEEP_SECONDS) - ); - } - - #[tokio::test(start_paused = true)] - async fn waits_for_requested_seconds() { - let handle = tokio::spawn(async { SleepTool::new().execute(json!({"seconds": 2})).await }); - tokio::task::yield_now().await; - tokio::time::advance(Duration::from_secs(1)).await; - tokio::task::yield_now().await; - assert!(!handle.is_finished()); - tokio::time::advance(Duration::from_secs(1)).await; - tokio::task::yield_now().await; - assert!(handle.await.unwrap().unwrap().success); - } - - #[tokio::test(start_paused = true)] - async fn waits_up_to_24_hour_boundary() { - let handle = tokio::spawn(async { - SleepTool::new() - .execute(json!({"seconds": MAX_SLEEP_SECONDS})) - .await - }); - tokio::task::yield_now().await; - tokio::time::advance(Duration::from_secs(MAX_SLEEP_SECONDS - 1)).await; - tokio::task::yield_now().await; - assert!(!handle.is_finished()); - tokio::time::advance(Duration::from_secs(1)).await; - tokio::task::yield_now().await; - assert!(handle.await.unwrap().unwrap().success); - } - - #[tokio::test(start_paused = true)] - async fn cancellation_drops_an_active_sleep() { - let handle = tokio::spawn(async { - SleepTool::new() - .execute(json!({"seconds": MAX_SLEEP_SECONDS})) - .await - }); - tokio::task::yield_now().await; - assert!(!handle.is_finished()); - handle.abort(); - assert!(handle.await.unwrap_err().is_cancelled()); - } - - #[tokio::test(start_paused = true)] - async fn user_cancellation_stops_sleep_and_terminalizes_its_tool_block() { - let (controller, emitter, receiver) = - TurnController::start("cli:test:sleep", "assistant-message"); - emitter - .emit(TurnEvent::ToolStarted { - iteration: 0, - call: ToolCall { - id: "sleep-call".into(), - name: "sleep".into(), - arguments: json!({"seconds": MAX_SLEEP_SECONDS}), - }, - }) - .unwrap(); - let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>(); - - let handle = tokio::spawn(async move { - let tool = SleepTool::new(); - tokio::select! { - result = tool.execute(json!({"seconds": MAX_SLEEP_SECONDS})) => { - result.unwrap(); - false - } - _ = cancel_rx => { - controller.cancel(Some("stopped by user".into())); - true - } - } - }); - tokio::task::yield_now().await; - drop(cancel_tx); - - assert!(handle.await.unwrap()); - let snapshot = receiver.borrow().clone(); - assert_eq!(snapshot.status, TurnStatus::Cancelled); - assert!(matches!( - &snapshot.blocks[0], - TurnBlock::Tool { - id, - status: ToolStatus::Cancelled, - .. - } if id == "sleep-call" - )); - } - - #[tokio::test(start_paused = true)] - async fn cancellation_token_ends_sleep_before_timer() { - let context = crate::tools::ToolExecutionContext::default(); - let token = context.cancellation.clone(); - let handle = tokio::spawn(async move { - SleepTool::new() - .execute_with_context(&context, json!({"seconds": MAX_SLEEP_SECONDS})) - .await - }); - tokio::task::yield_now().await; - assert!(!handle.is_finished()); - token.cancel(); - tokio::task::yield_now().await; - let error = handle.await.unwrap().unwrap_err(); - assert!(error.to_string().contains("cancelled")); - } - - #[tokio::test(start_paused = true)] - async fn pre_cancelled_context_never_enters_sleep() { - let context = crate::tools::ToolExecutionContext::default(); - context.cancellation.cancel(); - let error = SleepTool::new() - .execute_with_context(&context, json!({"seconds": 60})) - .await - .unwrap_err(); - assert!(error.to_string().contains("cancelled")); - } - - #[tokio::test(start_paused = true)] - async fn pending_input_before_listen_returns_immediately() { - let publisher = TurnWakeupPublisher::new(); - let handle = publisher.subscribe(); - publisher.publish(TurnWakeupState { - pending_user_steer: 1, - latest_source: Some(WakeupSource::UserSteer), - ..Default::default() - }); - let context = crate::tools::ToolExecutionContext::default().with_turn_wakeup(handle); - let result = SleepTool::new() - .execute_with_context(&context, json!({"seconds": 3600})) - .await - .unwrap(); - assert!(result.result.success); - assert!(result.result.output.contains("提前结束")); - assert!(result.result.output.contains("用户输入")); - } - - #[tokio::test(start_paused = true)] - async fn steer_publish_wakes_sleep_with_source_and_preview() { - let publisher = TurnWakeupPublisher::new(); - let handle = publisher.subscribe(); - let context = crate::tools::ToolExecutionContext::default().with_turn_wakeup(handle); - let tool = SleepTool::new(); - let wait = tokio::spawn(async move { - tool.execute_with_context(&context, json!({"seconds": 3600})) - .await - .unwrap() - .result - .output - }); - tokio::task::yield_now().await; - assert!(!wait.is_finished()); - publisher.publish(TurnWakeupState { - pending_agent_steer: 1, - latest_source: Some(WakeupSource::AgentSignal { - run_id: "run-123".to_string(), - agent_id: "monitor".to_string(), - }), - latest_safe_preview: Some("服务错误率超过 5%".to_string()), - ..Default::default() - }); - tokio::task::yield_now().await; - let output = wait.await.unwrap(); - assert!(output.contains("提前结束")); - assert!(output.contains("run-123")); - assert!(output.contains("服务错误率超过 5%")); - assert!(output.contains("安全边界注入")); - } - - #[tokio::test(start_paused = true)] - async fn queue_publish_wakes_sleep_without_content() { - let publisher = TurnWakeupPublisher::new(); - let handle = publisher.subscribe(); - let context = crate::tools::ToolExecutionContext::default().with_turn_wakeup(handle); - let tool = SleepTool::new(); - let wait = tokio::spawn(async move { - tool.execute_with_context(&context, json!({"seconds": 3600})) - .await - .unwrap() - .result - .output - }); - tokio::task::yield_now().await; - publisher.publish(TurnWakeupState { - pending_agent_queue: 1, - latest_source: Some(WakeupSource::AgentQueue), - ..Default::default() - }); - tokio::task::yield_now().await; - let output = wait.await.unwrap(); - assert!(output.contains("排队输入")); - assert!(output.contains("不会进入当前 Turn")); - assert!(!output.contains("run-")); - } - - #[tokio::test(start_paused = true)] - async fn child_sleep_without_handle_is_not_woken_by_publishes() { - let publisher = TurnWakeupPublisher::new(); - let _handle = publisher.subscribe(); - let context = crate::tools::ToolExecutionContext::default(); - let tool = SleepTool::new(); - let wait = tokio::spawn(async move { - tool.execute_with_context(&context, json!({"seconds": 30})) - .await - .unwrap() - .result - .output - }); - tokio::task::yield_now().await; - publisher.publish(TurnWakeupState { - pending_agent_steer: 1, - latest_source: Some(WakeupSource::AgentSignal { - run_id: "run-9".to_string(), - agent_id: "a".to_string(), - }), - ..Default::default() - }); - tokio::task::yield_now().await; - assert!(!wait.is_finished()); - tokio::time::advance(Duration::from_secs(30)).await; - tokio::task::yield_now().await; - assert!(wait.await.unwrap().contains("Slept for 30")); - } - - #[tokio::test(start_paused = true)] - async fn pre_listen_publish_does_not_lose_the_wake() { - // Publish BEFORE the sleep subscribes its own receiver: watch keeps - // the latest value, so the borrow_and_update pre-check sees it. - let publisher = TurnWakeupPublisher::new(); - let handle = publisher.subscribe(); - publisher.publish(TurnWakeupState { - pending_agent_queue: 2, - latest_source: Some(WakeupSource::AgentQueue), - ..Default::default() - }); - let context = crate::tools::ToolExecutionContext::default().with_turn_wakeup(handle); - let result = SleepTool::new() - .execute_with_context(&context, json!({"seconds": 3600})) - .await - .unwrap(); - assert!(result.result.success); - assert!(result.result.output.contains("排队输入")); - } -} diff --git a/src/tools/traits.rs b/src/tools/traits.rs index 7754555..dc9bd96 100644 --- a/src/tools/traits.rs +++ b/src/tools/traits.rs @@ -10,10 +10,6 @@ pub struct ToolExecutionContext { pub agent: Option>, pub cancellation: tokio_util::sync::CancellationToken, pub execution_gate: Option>, - /// Root interactive Turn only. Wake-aware tools (sleep) select on this - /// receiver so a user or Agent input ends the wait early; sub-runs and - /// continuations never receive it. - pub turn_wakeup: Option, } impl Default for ToolExecutionContext { @@ -24,7 +20,6 @@ impl Default for ToolExecutionContext { agent: None, cancellation: tokio_util::sync::CancellationToken::new(), execution_gate: None, - turn_wakeup: None, } } } @@ -37,7 +32,6 @@ impl ToolExecutionContext { agent: None, cancellation: tokio_util::sync::CancellationToken::new(), execution_gate: None, - turn_wakeup: None, } } @@ -66,18 +60,6 @@ impl ToolExecutionContext { self.execution_gate = Some(gate); self } - - pub fn with_turn_wakeup(mut self, handle: crate::agent::steering::TurnWakeupHandle) -> Self { - self.turn_wakeup = Some(handle); - self - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum InputInterruptPolicy { - Never, - WakeOnly, - CancelSafe, } #[derive(Debug, Clone)] @@ -221,11 +203,6 @@ pub trait Tool: Send + Sync + 'static { false } - /// Whether new Turn input may interrupt an in-flight invocation. - fn input_interrupt_policy(&self) -> InputInterruptPolicy { - InputInterruptPolicy::Never - } - /// Execute the tool through the unified output envelope. Most tools return /// only text and use this default conversion. async fn execute_output(&self, args: serde_json::Value) -> anyhow::Result {