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.
This commit is contained in:
parent
b2574dc7af
commit
b13450498b
@ -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
|
||||
|
||||
|
||||
@ -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)。
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -9,7 +9,6 @@ tools:
|
||||
- content_search
|
||||
- web_fetch
|
||||
- calculator
|
||||
- sleep
|
||||
---
|
||||
|
||||
# General Purpose Agent
|
||||
|
||||
@ -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 风险。
|
||||
|
||||
@ -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<WakeupSource>,
|
||||
/// Safe, model-visible preview for steer wakes only. Queue wakes never
|
||||
/// carry content.
|
||||
pub latest_safe_preview: Option<String>,
|
||||
}
|
||||
|
||||
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<TurnWakeupState>,
|
||||
}
|
||||
|
||||
/// 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<TurnWakeupState>,
|
||||
}
|
||||
|
||||
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)]
|
||||
|
||||
@ -663,8 +663,6 @@ struct ActiveTurnEmitter {
|
||||
/// `AgentTurnContext`; keeping it on the session handle makes admission
|
||||
/// atomic with `/stop` and worker cleanup.
|
||||
steering: Arc<TurnMailbox>,
|
||||
/// 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<String>,
|
||||
) {
|
||||
) -> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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"
|
||||
));
|
||||
}
|
||||
|
||||
|
||||
@ -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<ToolRegistry> {
|
||||
let registry = ToolRegistry::new();
|
||||
registry.register(CalculatorTool::new());
|
||||
registry.register(SleepTool::new());
|
||||
registry.register(FileReadTool::new());
|
||||
registry.register(FileWriteTool::new());
|
||||
registry.register(FileEditTool::new());
|
||||
|
||||
@ -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<u64, String> {
|
||||
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<ToolResult> {
|
||||
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<crate::tools::ToolOutput> {
|
||||
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("排队输入"));
|
||||
}
|
||||
}
|
||||
@ -10,10 +10,6 @@ pub struct ToolExecutionContext {
|
||||
pub agent: Option<std::sync::Arc<crate::agent::AgentExecutionContext>>,
|
||||
pub cancellation: tokio_util::sync::CancellationToken,
|
||||
pub execution_gate: Option<std::sync::Arc<crate::agent::gate::ExecutionGate>>,
|
||||
/// 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<crate::agent::steering::TurnWakeupHandle>,
|
||||
}
|
||||
|
||||
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<ToolOutput> {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user