From 554d4b15acf39b06c58bf077d377b8f0133bf32f Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Tue, 11 Aug 2026 15:24:49 +0800 Subject: [PATCH] feat: wire run quota and signal delivery persistence; remove redundant compat code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RunQuota now enforced: background admission acquires the run permit (global→session) and the runner holds it until terminal commit; foreground never takes run permits so nested limit=1 cannot deadlock - RuntimeAdmission activity guard held by background runners for the whole run; admission checked before any durable write - signal_contract_json/signal_delivery persisted at run admission so definition-level steer delivery actually takes effect (was silently falling back to queue) - terminal completion payload carries the run's emitted signal IDs (design §12.3) so the main Agent can recognise duplicates - ROOT caller_scope_id is the literal "ROOT" (design §9.6) - legacy general delegation returns a deprecation/migration hint Cleanup: - remove unused AgentCaller enum and run_parallel wrapper - remove unreachable http_get_only prompt machinery (http_request is RootOnly and never enters sub-agent registries) - drop never-read EmittedSignal fields; derive WakeupSource from TurnInputSource instead of duplicating match arms - replace the /stop oneshot compatibility bridge with a plain turn_busy flag; the worker's forced-cancel path now selects on the cancellation token (implementation doc §11.1) Version 1.8.0 --- Cargo.toml | 2 +- .../SUB_AGENT_ORCHESTRATION_IMPLEMENTATION.md | 4 +- .../skills/about-picobot/references/tools.md | 2 + src/agent/coordinator.rs | 275 ++++++++++++++- src/agent/mod.rs | 2 +- src/agent/run.rs | 8 - src/agent/steering.rs | 126 +++++++ src/agent/sub_agent.rs | 20 +- src/agent/system_prompt.rs | 16 +- src/session/session.rs | 327 +++++++++++++++--- src/storage/agent_inbox.rs | 6 + src/storage/agent_run.rs | 51 ++- src/tools/delegate.rs | 19 +- src/tools/emit_signal.rs | 2 - src/tools/sleep.rs | 249 ++++++++++++- src/tools/traits.rs | 11 + webui/package-lock.json | 4 +- webui/package.json | 2 +- 18 files changed, 1003 insertions(+), 123 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index c2439be..5d09491 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "picobot" -version = "1.7.0" +version = "1.8.0" edition = "2024" [dependencies] diff --git a/docs/SUB_AGENT_ORCHESTRATION_IMPLEMENTATION.md b/docs/SUB_AGENT_ORCHESTRATION_IMPLEMENTATION.md index 4d806b3..03db8d0 100644 --- a/docs/SUB_AGENT_ORCHESTRATION_IMPLEMENTATION.md +++ b/docs/SUB_AGENT_ORCHESTRATION_IMPLEMENTATION.md @@ -6,7 +6,9 @@ > > 本文不是“代码已经实现”的声明。实现完成前，运行时事实仍以当前代码、测试和 [`ARCHITECTURE.md`](ARCHITECTURE.md) 为准。 -> 实施进度（2026-08）：Phase 1 已落地具名 Definition/Catalog、不同 Provider profile、工具/Skill fail-closed 裁剪、显式 `AgentExecutionContext`、父子委托边与 ancestry 校验、canonical `foreground/background` schema 以及批量 foreground 并发。旧 general background 仅作为兼容路径保留。Phase 2A 已落地结构化取消：`AgentError::Cancelled/TimedOut`、CancellationToken 贯穿 root Turn、Provider 连接/stream、并行与串行工具批次以及 sleep；`/stop` 保留 oneshot 兼容桥接并同时取消 Turn token，协作式与强制路径提交同一 Cancelled 终态；树级 `max_runs_per_tree` 由共享原子计数在 foreground 委托接纳时强制。Phase 2B 已落地 durable foreground 编排：schema v6（`agent_run_groups`/`agent_runs`/`agent_session_state`/`agent_inbox_events` 与 messages/sessions 扩展列）在单一迁移事务中原子创建；Storage 领域 API（接纳、running/waiting_children 条件转换、execution-ID 条件 terminal commit、plan item 原子领取/完成、游标分页）；`ExecutionGate` 分离 run quota 与 provider/tool step gate（global→session 顺序获取、弱引用键控回收、取消可中断等待），root Turn 步骤同样占用 step gate；`AgentCoordinator` 持久化全部具名 foreground run（单任务不建 group、批量建 group 并按请求顺序返回）、父 run `waiting_children` 转换、迟到结果丢弃与树位置授权；`agent_task` 工具提供 scoped get/list/get_result/cancel。foreground 不占用 run quota，嵌套 foreground 在并发上限为 1 时不死锁。Phase 3 已落地 durable inbox 与 queue continuation：具名 background 单任务经 Coordinator 接纳（completion slot 预留 → 持久化 queued → TaskSupervisor 托管 runner → terminal commit 原子转换 reservation 为 completion 事件 → notifier wake），spawn 拒绝执行补偿事务；`agent_inbox_events`/`agent_session_state` 容量条件更新与 claim/lease/admit/release/supersede/dead-letter API；`AgentInboxNotifier`（弱引用 late-bound wake）与 Session 双 lane worker（user mpsc + inbox watch 合并 wake），公平调度按 `max_user_turn_burst_before_inbox` 与 `max_inbox_wait_secs` 强制 continuation；continuation Turn 使用 hidden trigger、只读工具集，`commit_continuation_turn` 同事务提交 hidden trigger + assistant/tool/usage + event consume，失败显式 release lease；`client_visibility`/`turn_origin` 贯穿 ChatMessage/MessageMeta/协议 DTO，客户端历史查询默认过滤 hidden，`message_count` 只统计可见用户输入；activation recovery 收敛旧代 run（interrupted + failure completion）、过期 lease、group counter 与容量计数；`/stop`/archive/delete 走 `cancel_session`（suppress_continuation 写 consumed completion）并 dead-letter。background completion 不再直接通知 Channel。Phase 3 审查修复：worker 通过 `next_pending_due_at` 定时器在 release backoff 到期后重新 claim（不再依赖 wake）；达到 `max_inbox_delivery_attempts` 的事件在正常运行中即 dead-letter 而非无限重试；`recover_on_activation` 对含 due 事件的在内存 session 发送合并 wake；修正 MIN 聚合无行时 NULL 被解码为 0 导致 worker 空转的缺陷（`oldest_pending_due`/`next_pending_due_at` 用 `Option>` 显式解码）。Phase 4 已落地 emit_signal 与 steer：Agent Definition frontmatter 新增 `signal:` 块（`SignalContract`：delivery queue/steer、总数/字节/间隔/burst/severity allowlist/dedupe 冷却窗/JSON 深度，全部由工具与 Coordinator 强制，模型只提供 key/severity/summary/details/dedupe_key）；`EmitSignalTool` 仅在带 contract 的 run 注册（fail-closed），`insert_agent_signal` 支持同冷却窗 dedupe（返回原 event + deduplicated）；Coordinator `emit_signal` 校验 run 活跃与 execution ID、容量条件插入、投影+notifier wake，取消 run 时未消费 signal 自动 supersede；SteeringMailbox 泛化为来源感知 TurnMailbox（user lane 32/64KiB 与 agent lane 8/32KiB 独立容量，`TurnInput{source,delivery,durable_event_id,lease_token}`，agent steer 投影为 hidden user 消息保留 source 元数据）；steer 两阶段 admission（claim → mailbox 预留 → DB admit(turn_id) → 同 Turn/generation 激活），任何失败 release lease 并 wake queue lane；`/stop`/generation 变更时已 admit 的 steer 事件按 lease token 条件释放回 pending（绝不静默丢弃），用户 Turn commit 与 steer 事件 consume 同事务（`persist_turn_batch_with_steer_consumption`）。Phase 3 收尾已落地：`ChannelContext.durable_private`（Feishu 仅 thread/root/chat_type 进入）持久化到 `sessions.delivery_context` 并被 continuation 投递复用；WS 协议 `GetAgentRuns`/`GetAgentRun` 与 `SessionAgentRuns`/`AgentRunUpdated`/`AgentEventUpdated`（有界 `AgentRunView`/`AgentEventView`，不暴露 budget/contract/delivery context/execution id）；`AgentProjectionHub` broadcast（复用 plan-change 模式，lag 由客户端 GetAgentRuns 校准）；HTTP `/api/agent-runs*`（列表游标分页/详情/events/cancel）+ `/api/tasks` legacy/new union；WebUI TasksPage 后台 tab 渲染 run tree（group 折叠、agent 标签、深度缩进），ChatPage 显示 continuation 标签与 Signal 卡片（投影事件，不插入 history）。Phase 5 尚未完成：wake-aware sleep 与工具中断策略。 +> 实施进度（2026-08）：Phase 1 已落地具名 Definition/Catalog、不同 Provider profile、工具/Skill fail-closed 裁剪、显式 `AgentExecutionContext`、父子委托边与 ancestry 校验、canonical `foreground/background` schema 以及批量 foreground 并发。旧 general background 仅作为兼容路径保留。Phase 2A 已落地结构化取消：`AgentError::Cancelled/TimedOut`、CancellationToken 贯穿 root Turn、Provider 连接/stream、并行与串行工具批次以及 sleep；`/stop` 保留 oneshot 兼容桥接并同时取消 Turn token，协作式与强制路径提交同一 Cancelled 终态；树级 `max_runs_per_tree` 由共享原子计数在 foreground 委托接纳时强制。Phase 2B 已落地 durable foreground 编排：schema v6（`agent_run_groups`/`agent_runs`/`agent_session_state`/`agent_inbox_events` 与 messages/sessions 扩展列）在单一迁移事务中原子创建；Storage 领域 API（接纳、running/waiting_children 条件转换、execution-ID 条件 terminal commit、plan item 原子领取/完成、游标分页）；`ExecutionGate` 分离 run quota 与 provider/tool step gate（global→session 顺序获取、弱引用键控回收、取消可中断等待），root Turn 步骤同样占用 step gate；`AgentCoordinator` 持久化全部具名 foreground run（单任务不建 group、批量建 group 并按请求顺序返回）、父 run `waiting_children` 转换、迟到结果丢弃与树位置授权；`agent_task` 工具提供 scoped get/list/get_result/cancel。foreground 不占用 run quota，嵌套 foreground 在并发上限为 1 时不死锁。Phase 3 已落地 durable inbox 与 queue continuation：具名 background 单任务经 Coordinator 接纳（completion slot 预留 → 持久化 queued → TaskSupervisor 托管 runner → terminal commit 原子转换 reservation 为 completion 事件 → notifier wake），spawn 拒绝执行补偿事务；`agent_inbox_events`/`agent_session_state` 容量条件更新与 claim/lease/admit/release/supersede/dead-letter API；`AgentInboxNotifier`（弱引用 late-bound wake）与 Session 双 lane worker（user mpsc + inbox watch 合并 wake），公平调度按 `max_user_turn_burst_before_inbox` 与 `max_inbox_wait_secs` 强制 continuation；continuation Turn 使用 hidden trigger、只读工具集，`commit_continuation_turn` 同事务提交 hidden trigger + assistant/tool/usage + event consume，失败显式 release lease；`client_visibility`/`turn_origin` 贯穿 ChatMessage/MessageMeta/协议 DTO，客户端历史查询默认过滤 hidden，`message_count` 只统计可见用户输入；activation recovery 收敛旧代 run（interrupted + failure completion）、过期 lease、group counter 与容量计数；`/stop`/archive/delete 走 `cancel_session`（suppress_continuation 写 consumed completion）并 dead-letter。background completion 不再直接通知 Channel。Phase 3 审查修复：worker 通过 `next_pending_due_at` 定时器在 release backoff 到期后重新 claim（不再依赖 wake）；达到 `max_inbox_delivery_attempts` 的事件在正常运行中即 dead-letter 而非无限重试；`recover_on_activation` 对含 due 事件的在内存 session 发送合并 wake；修正 MIN 聚合无行时 NULL 被解码为 0 导致 worker 空转的缺陷（`oldest_pending_due`/`next_pending_due_at` 用 `Option>` 显式解码）。Phase 4 已落地 emit_signal 与 steer：Agent Definition frontmatter 新增 `signal:` 块（`SignalContract`：delivery queue/steer、总数/字节/间隔/burst/severity allowlist/dedupe 冷却窗/JSON 深度，全部由工具与 Coordinator 强制，模型只提供 key/severity/summary/details/dedupe_key）；`EmitSignalTool` 仅在带 contract 的 run 注册（fail-closed），`insert_agent_signal` 支持同冷却窗 dedupe（返回原 event + deduplicated）；Coordinator `emit_signal` 校验 run 活跃与 execution ID、容量条件插入、投影+notifier wake，取消 run 时未消费 signal 自动 supersede；SteeringMailbox 泛化为来源感知 TurnMailbox（user lane 32/64KiB 与 agent lane 8/32KiB 独立容量，`TurnInput{source,delivery,durable_event_id,lease_token}`，agent steer 投影为 hidden user 消息保留 source 元数据）；steer 两阶段 admission（claim → mailbox 预留 → DB admit(turn_id) → 同 Turn/generation 激活），任何失败 release lease 并 wake queue lane；`/stop`/generation 变更时已 admit 的 steer 事件按 lease token 条件释放回 pending（绝不静默丢弃），用户 Turn commit 与 steer 事件 consume 同事务（`persist_turn_batch_with_steer_consumption`）。Phase 3 收尾已落地：`ChannelContext.durable_private`（Feishu 仅 thread/root/chat_type 进入）持久化到 `sessions.delivery_context` 并被 continuation 投递复用；WS 协议 `GetAgentRuns`/`GetAgentRun` 与 `SessionAgentRuns`/`AgentRunUpdated`/`AgentEventUpdated`（有界 `AgentRunView`/`AgentEventView`，不暴露 budget/contract/delivery context/execution id）；`AgentProjectionHub` broadcast（复用 plan-change 模式，lag 由客户端 GetAgentRuns 校准）；HTTP `/api/agent-runs*`（列表游标分页/详情/events/cancel）+ `/api/tasks` legacy/new union；WebUI TasksPage 后台 tab 渲染 run tree（group 折叠、agent 标签、深度缩进），ChatPage 显示 continuation 标签与 Signal 卡片（投影事件，不插入 history）。Phase 5 已落地 wake-aware sleep：`TurnWakeupPublisher`/`TurnWakeupHandle` 随 active Turn 创建（root interactive Turn 专用，sub-run 与 continuation 永远没有 handle）；watch revision 单调，四个 admission 点（用户进 TurnMailbox、用户进 next-turn mpsc、Agent steer durable admitted、Agent event 保持 pending 且 wake Session）在事实可见后 publish（先更新状态再发 wake，sleep 醒来必能查到输入）；`SleepTool` root 上下文先 `borrow_and_update` 预检 pending>0 立即返回，否则 select timer/changed/cancellation，steer 唤醒消息携带 run_id/agent_id 与安全摘要、queue 唤醒只给类型/数量并明确不泄漏正文；child/continuation sleep 仅 timer/cancel（不受 root 输入或 sibling signal 影响）。`InputInterruptPolicy` 元数据沿用（sleep=WakeOnly，其余默认 Never，不扩展）。legacy general background 的 direct notification 兼容路径按设计保留一个版本观察，随后删除旧 adapter 与 `background_tasks` 写入。 + +三方对齐审查（2026-08）修复：RunQuota 接线（background admission 前按 global→session 获取 run permit，随 runner 持有至 terminal commit；foreground 不占 run permit，嵌套 limit=1 永不死锁，与 gate 注释语义一致）；`signal_contract_json`/`signal_delivery` 在 run 接纳时持久化（此前 steer delivery 静默回退为 queue）；terminal completion payload 携带本 run 已发出的 signal IDs（§12.3）；ROOT 的 `caller_scope_id` 固定为字面量 `"ROOT"`（§9.6）；legacy general 委托返回弃用迁移提示（§4.3）。已知剩余偏差：`resource_scope_id` 未建模为显式字段（browser 瞬态隔离经 session_id 惯例达成，行为等效）；`idempotency_key` 仅 schema 预留、工具未开放入口；`all` policy 的 group completion 事件在 background 批量开放前不可达；legacy 适配器与 `background_tasks` 写入按设计保留一个版本后删除。 ## 1. 审查结论 diff --git a/resources/skills/about-picobot/references/tools.md b/resources/skills/about-picobot/references/tools.md index 7beefa6..1ac8efb 100644 --- a/resources/skills/about-picobot/references/tools.md +++ b/resources/skills/about-picobot/references/tools.md @@ -219,6 +219,8 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 参数 `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/coordinator.rs b/src/agent/coordinator.rs index 81bb855..5ca66ce 100644 --- a/src/agent/coordinator.rs +++ b/src/agent/coordinator.rs @@ -32,6 +32,8 @@ pub struct AgentCoordinator { work_manager: Option>, notifier: Arc, projection: Arc, + execution_gate: Arc, + admission: crate::gateway::reload::RuntimeAdmission, task_supervisor: crate::task_supervisor::TaskSupervisor, runtime_generation: i64, max_pending_inbox_events_per_session: i64, @@ -51,12 +53,14 @@ pub enum CoordinatorError { impl AgentCoordinator { #[allow(clippy::too_many_arguments)] - pub fn new( + pub(crate) fn new( storage: Arc, manager: Arc, work_manager: Arc, notifier: Arc, projection: Arc, + execution_gate: Arc, + admission: crate::gateway::reload::RuntimeAdmission, task_supervisor: crate::task_supervisor::TaskSupervisor, runtime_generation: u64, orchestration: &crate::config::AgentOrchestrationConfig, @@ -67,6 +71,8 @@ impl AgentCoordinator { work_manager: Some(work_manager), notifier, projection, + execution_gate, + admission, task_supervisor, runtime_generation: runtime_generation as i64, max_pending_inbox_events_per_session: orchestration.max_pending_inbox_events_per_session @@ -110,6 +116,21 @@ impl AgentCoordinator { })?; let now = chrono::Utc::now().timestamp_millis(); + // 0. Run quota + admission guard before any durable write; any + // failure here releases everything without touching SQLite. The + // permit stays with the runner until the terminal commit. + let run_permit = self + .execution_gate + .acquire_run(&root_session_id, &caller.cancellation) + .await + .map_err(|error| CoordinatorError::Rejected(error.to_string()))?; + let activity = self.admission.try_enter().ok_or_else(|| { + CoordinatorError::Rejected( + "gateway is draining for configuration reload and cannot accept background tasks" + .to_string(), + ) + })?; + // 1. Reserve the completion slot; failure means the inbox is full and // nothing is admitted. if self @@ -135,7 +156,7 @@ impl AgentCoordinator { root_turn_id: caller.turn_id.clone(), parent_run_id: None, caller_agent_id: "ROOT".to_string(), - caller_scope_id: caller.turn_id.clone().unwrap_or_else(|| "root".to_string()), + caller_scope_id: "ROOT".to_string(), idempotency_key: None, agent_id: resolved.agent_id.clone().unwrap_or_default(), definition_hash: resolved.definition_hash.clone().unwrap_or_default(), @@ -153,6 +174,14 @@ impl AgentCoordinator { "remaining_depth": self.manager.catalog().max_tree_depth(), }) .to_string(), + signal_contract_json: resolved + .signal_contract + .as_ref() + .map(|contract| serde_json::to_string(contract).unwrap_or_default()), + signal_delivery: resolved + .signal_contract + .as_ref() + .map(|contract| contract.delivery.as_str().to_string()), deadline_at: now + (resolved.timeout_secs * 1000) as i64, runtime_generation: self.runtime_generation, completion_slot_reserved: true, @@ -190,12 +219,16 @@ impl AgentCoordinator { let run_id = run_id.clone(); async move { coordinator - .run_background_runner(&run_id, &config, resolved, token) + .run_background_runner( + &run_id, &config, resolved, token, run_permit, activity, + ) .await; } }); if !spawned { // Compensation: undo the durable admission before returning. + // The rejected closure was dropped by the supervisor, which + // released the run quota permit and activity guard. self.active_tokens.remove(&run_id); let _ = self .storage @@ -214,6 +247,8 @@ impl AgentCoordinator { config: &SubAgentConfig, resolved: ResolvedAgentRun, token: CancellationToken, + _run_permit: crate::agent::gate::RunPermit, + _activity: crate::gateway::reload::ActivityGuard, ) { let now = chrono::Utc::now().timestamp_millis(); let execution_id = run_id.to_string(); @@ -229,6 +264,19 @@ impl AgentCoordinator { return; } + let emitted_signals: Vec = resolved + .tool_context + .agent + .as_ref() + .and_then(|agent| agent.emitted_signals.lock().ok()) + .map(|signals| { + signals + .iter() + .map(|signal| signal.signal_id.clone()) + .collect() + }) + .unwrap_or_default(); + let result = self .manager .execute_resolved(config, resolved, run_id) @@ -243,18 +291,22 @@ impl AgentCoordinator { cost: None, tool_calls: result.tool_calls_count as i64, iterations: result.iterations as i64, + signal_ids: emitted_signals.clone(), }, TaskStatus::Failed(error) => AgentTerminalOutcome::Failed { error: error.clone(), prompt_tokens: None, completion_tokens: None, cost: None, + signal_ids: emitted_signals.clone(), }, TaskStatus::TimedOut => AgentTerminalOutcome::TimedOut { deadline_at: chrono::Utc::now().timestamp_millis(), + signal_ids: emitted_signals.clone(), }, TaskStatus::Cancelled => AgentTerminalOutcome::Cancelled { reason: "cancelled by user, parent or shutdown".to_string(), + signal_ids: emitted_signals.clone(), }, }, Err(error) => AgentTerminalOutcome::Failed { @@ -262,6 +314,7 @@ impl AgentCoordinator { prompt_tokens: None, completion_tokens: None, cost: None, + signal_ids: emitted_signals, }, }; @@ -478,6 +531,14 @@ impl AgentCoordinator { "remaining_depth": caller.agent.as_ref().map(|agent| agent.budget.remaining_depth), })) .unwrap_or_default(), + signal_contract_json: resolution + .signal_contract + .as_ref() + .map(|contract| serde_json::to_string(contract).unwrap_or_default()), + signal_delivery: resolution + .signal_contract + .as_ref() + .map(|contract| contract.delivery.as_str().to_string()), deadline_at, runtime_generation: self.runtime_generation, completion_slot_reserved: false, @@ -596,6 +657,19 @@ impl AgentCoordinator { }); } + let emitted_signals: Vec = resolution + .tool_context + .agent + .as_ref() + .and_then(|agent| agent.emitted_signals.lock().ok()) + .map(|signals| { + signals + .iter() + .map(|signal| signal.signal_id.clone()) + .collect() + }) + .unwrap_or_default(); + let result = self .manager .execute_resolved(config, resolution, run_id) @@ -610,18 +684,22 @@ impl AgentCoordinator { cost: None, tool_calls: result.tool_calls_count as i64, iterations: result.iterations as i64, + signal_ids: emitted_signals.clone(), }, TaskStatus::Failed(error) => AgentTerminalOutcome::Failed { error: error.clone(), prompt_tokens: None, completion_tokens: None, cost: None, + signal_ids: emitted_signals.clone(), }, TaskStatus::TimedOut => AgentTerminalOutcome::TimedOut { deadline_at: chrono::Utc::now().timestamp_millis(), + signal_ids: emitted_signals.clone(), }, TaskStatus::Cancelled => AgentTerminalOutcome::Cancelled { reason: "cancelled by user, parent or shutdown".to_string(), + signal_ids: emitted_signals.clone(), }, }, Err(error) => AgentTerminalOutcome::Failed { @@ -629,6 +707,7 @@ impl AgentCoordinator { prompt_tokens: None, completion_tokens: None, cost: None, + signal_ids: emitted_signals, }, }; @@ -1038,6 +1117,12 @@ mod tests { } async fn coordinator() -> (Arc, tempfile::TempDir) { + coordinator_with_inbox_limit(1).await + } + + async fn coordinator_with_inbox_limit( + max_pending: usize, + ) -> (Arc, tempfile::TempDir) { let dir = tempfile::tempdir().unwrap(); let storage = Arc::new(Storage::new(&dir.path().join("coord.db")).await.unwrap()); let (notify_tx, _notify_rx) = tokio::sync::mpsc::unbounded_channel(); @@ -1059,7 +1144,7 @@ mod tests { let supervisor = crate::task_supervisor::TaskSupervisor::new(); let orchestration = crate::config::AgentOrchestrationConfig { enabled: true, - max_pending_inbox_events_per_session: 1, + max_pending_inbox_events_per_session: max_pending, ..Default::default() }; ( @@ -1069,6 +1154,8 @@ mod tests { work_manager, notifier, Arc::new(crate::agent::AgentProjectionHub::new()), + crate::agent::gate::ExecutionGate::unbounded(), + crate::gateway::reload::RuntimeAdmission::open(), supervisor, 1, &orchestration, @@ -1259,12 +1346,13 @@ mod tests { assert_eq!(state, (1, 0)); } - async fn storage_accept_run( + async fn storage_accept_run_with_delivery( storage: &Arc, run_id: &str, session: &str, now: i64, slot_reserved: bool, + signal_delivery: Option<&str>, ) { let _ = storage.ensure_agent_session_state(session, now).await; if slot_reserved { @@ -1290,6 +1378,8 @@ mod tests { task: "work".to_string(), context_json: None, budget_json: "{}".to_string(), + signal_contract_json: signal_delivery.map(|_| "{}".to_string()), + signal_delivery: signal_delivery.map(str::to_string), deadline_at: now + 100_000, runtime_generation: 1, completion_slot_reserved: slot_reserved, @@ -1304,6 +1394,16 @@ mod tests { .unwrap(); } + async fn storage_accept_run( + storage: &Arc, + run_id: &str, + session: &str, + now: i64, + slot_reserved: bool, + ) { + storage_accept_run_with_delivery(storage, run_id, session, now, slot_reserved, None).await + } + #[tokio::test] async fn emit_signal_persists_wakes_and_respects_capacity_and_dedupe() { let (coordinator, _dir) = coordinator().await; @@ -1373,6 +1473,98 @@ mod tests { )); } + #[tokio::test] + async fn terminal_commit_carries_emitted_signal_ids_in_completion_payload() { + let (coordinator, _dir) = coordinator_with_inbox_limit(8).await; + let run_id = "run-sig-carrier"; + let session = "cli:test:dialog"; + let now = chrono::Utc::now().timestamp_millis(); + storage_accept_run(&coordinator.storage, run_id, session, now, true).await; + + // Emit one signal, then commit the run terminal. + let context = crate::agent::AgentExecutionContext { + root_session_id: session.to_string(), + root_turn_id: None, + run_id: run_id.to_string(), + execution_id: run_id.to_string(), + group_id: None, + parent_run_id: None, + caller_agent_id: "ROOT".to_string(), + current_agent_id: "researcher".to_string(), + ancestry: vec!["researcher".to_string()], + depth: 1, + plan_item_id: None, + cancellation: tokio_util::sync::CancellationToken::new(), + budget: crate::agent::AgentBudget { + remaining_runs: 15, + remaining_depth: 3, + }, + tree_runs: Arc::new(std::sync::atomic::AtomicUsize::new(1)), + signal_contract: None, + emitted_signals: Arc::new(std::sync::Mutex::new(Vec::new())), + }; + let signal = SignalInput { + key: "k".to_string(), + severity: "warning".to_string(), + summary: "s".to_string(), + details: None, + dedupe_key: None, + event_key: format!("signal:{}", uuid::Uuid::new_v4()), + }; + let accepted = coordinator.emit_signal(&context, signal).await.unwrap(); + context + .emitted_signals + .lock() + .unwrap() + .push(crate::agent::run::EmittedSignal { + signal_id: accepted.signal_id.clone(), + }); + + let signal_ids: Vec = context + .emitted_signals + .lock() + .unwrap() + .iter() + .map(|signal| signal.signal_id.clone()) + .collect(); + coordinator + .storage + .commit_agent_terminal( + run_id, + run_id, + 1, + &AgentTerminalOutcome::Completed { + result: "done".to_string(), + prompt_tokens: None, + completion_tokens: None, + cost: None, + tool_calls: 1, + iterations: 1, + signal_ids, + }, + None, + now + 1, + ) + .await + .unwrap(); + + let events = coordinator + .storage + .list_agent_inbox_events(session, 10) + .await + .unwrap(); + assert_eq!(events.len(), 2); + let completion = events + .iter() + .find(|event| { + event.event_type == crate::storage::agent_inbox::AgentEventType::Completion + }) + .unwrap(); + let payload: serde_json::Value = serde_json::from_str(&completion.payload_json).unwrap(); + assert_eq!(payload["status"], "completed"); + assert_eq!(payload["signal_ids"][0], accepted.signal_id); + } + #[tokio::test] async fn emit_signal_rejects_when_run_is_terminal_or_inbox_is_full() { let (coordinator, _dir) = coordinator().await; @@ -1419,6 +1611,7 @@ mod tests { 1, &AgentTerminalOutcome::Cancelled { reason: "test".to_string(), + signal_ids: Vec::new(), }, None, now + 1, @@ -1462,6 +1655,8 @@ mod tests { work_manager, notifier, Arc::new(crate::agent::AgentProjectionHub::new()), + crate::agent::gate::ExecutionGate::unbounded(), + crate::gateway::reload::RuntimeAdmission::open(), supervisor, 1, &orchestration, @@ -1494,4 +1689,74 @@ mod tests { Err(CoordinatorError::Rejected(_)) )); } + + #[tokio::test] + async fn emitted_signal_uses_the_runs_persisted_delivery_lane() { + let (coordinator, _dir) = coordinator_with_inbox_limit(8).await; + let run_id = "run-sig-steer"; + let session = "cli:test:dialog"; + let now = chrono::Utc::now().timestamp_millis(); + storage_accept_run_with_delivery( + &coordinator.storage, + run_id, + session, + now, + false, + Some("steer"), + ) + .await; + + let run = coordinator + .storage + .get_agent_run(run_id) + .await + .unwrap() + .unwrap(); + assert_eq!(run.signal_delivery.as_deref(), Some("steer")); + + let context = crate::agent::AgentExecutionContext { + root_session_id: session.to_string(), + root_turn_id: None, + run_id: run_id.to_string(), + execution_id: run_id.to_string(), + group_id: None, + parent_run_id: None, + caller_agent_id: "ROOT".to_string(), + current_agent_id: "researcher".to_string(), + ancestry: vec!["researcher".to_string()], + depth: 1, + plan_item_id: None, + cancellation: tokio_util::sync::CancellationToken::new(), + budget: crate::agent::AgentBudget { + remaining_runs: 15, + remaining_depth: 3, + }, + tree_runs: Arc::new(std::sync::atomic::AtomicUsize::new(1)), + signal_contract: None, + emitted_signals: Arc::new(std::sync::Mutex::new(Vec::new())), + }; + let signal = SignalInput { + key: "k".to_string(), + severity: "info".to_string(), + summary: "s".to_string(), + details: None, + dedupe_key: None, + event_key: format!("signal:{}", uuid::Uuid::new_v4()), + }; + let accepted = coordinator.emit_signal(&context, signal).await.unwrap(); + assert!(matches!( + accepted.delivery, + crate::storage::agent_inbox::AgentEventDelivery::Steer + )); + let event = coordinator + .storage + .get_agent_inbox_event(&accepted.signal_id) + .await + .unwrap() + .unwrap(); + assert_eq!( + event.delivery, + crate::storage::agent_inbox::AgentEventDelivery::Steer + ); + } } diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 2b7e504..961ae5b 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -21,7 +21,7 @@ pub use definition::{AgentDefinition, AgentLimits}; pub use gate::ExecutionGate; pub use inbox::{AgentInboxNotifier, AgentInboxWakeTarget}; pub use projection::AgentProjectionHub; -pub use run::{AgentBudget, AgentCaller, AgentExecutionContext}; +pub use run::{AgentBudget, AgentExecutionContext}; pub use steering::{SteeringDrain, SteeringPushError, TurnInput, TurnInputSource, TurnMailbox}; pub use sub_agent::{ DelegateContext, ExecutionMode, SubAgentConfig, SubAgentError, SubAgentManager, SubAgentResult, diff --git a/src/agent/run.rs b/src/agent/run.rs index c1ce8f8..4901499 100644 --- a/src/agent/run.rs +++ b/src/agent/run.rs @@ -4,12 +4,6 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use tokio_util::sync::CancellationToken; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AgentCaller { - Root, - Agent, -} - #[derive(Debug, Clone)] pub struct AgentBudget { pub remaining_runs: usize, @@ -21,8 +15,6 @@ pub struct AgentBudget { #[derive(Debug, Clone)] pub struct EmittedSignal { pub signal_id: String, - pub severity: String, - pub summary: String, } #[derive(Debug, Clone)] diff --git a/src/agent/steering.rs b/src/agent/steering.rs index 0aeeba5..5c37738 100644 --- a/src/agent/steering.rs +++ b/src/agent/steering.rs @@ -47,6 +47,104 @@ 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(), + } + } + TurnInputSource::AgentGroupCompletion { group_id } => { + WakeupSource::AgentGroupCompletion { + group_id: group_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 }, + AgentGroupCompletion { group_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)] @@ -566,6 +664,34 @@ impl TurnMailbox { pub fn max_agent_messages(&self) -> usize { self.max_agent_messages } + + /// Currently pending user steer entries (wake-state hint). + pub fn user_pending_count(&self) -> usize { + let state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + state + .pending + .iter() + .filter(|input| !input.source.is_agent()) + .count() + } + + /// Currently pending agent steer entries, including reservations + /// (wake-state hint). + pub fn agent_pending_count(&self) -> usize { + let state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + state + .pending + .iter() + .chain(state.reserved.iter()) + .filter(|input| input.source.is_agent()) + .count() + } } impl Default for TurnMailbox { diff --git a/src/agent/sub_agent.rs b/src/agent/sub_agent.rs index 476f01e..ff78e1a 100644 --- a/src/agent/sub_agent.rs +++ b/src/agent/sub_agent.rs @@ -154,6 +154,9 @@ pub(crate) struct ResolvedAgentRun { pub agent_id: Option, pub definition_hash: Option, pub llm_profile: Option, + /// Durable signal contract of the definition; `None` means the run can + /// never emit signals and must not persist contract state. + pub signal_contract: Option, } impl SubAgentManager { @@ -293,6 +296,7 @@ impl SubAgentManager { agent_id: None, definition_hash: None, llm_profile: None, + signal_contract: None, }); }; @@ -469,6 +473,7 @@ impl SubAgentManager { agent_id: Some(target.to_string()), definition_hash: Some(definition.definition_hash.clone()), llm_profile: Some(definition.llm_profile.clone()), + signal_contract: definition.signal_contract.clone(), tool_context: ToolExecutionContext::for_session(format!("agent-run:{task_id}")) .with_turn_id( caller @@ -564,7 +569,6 @@ impl SubAgentManager { &resolved.provider_config.workspace_dir, &resolved.provider_config.model_id, resolved.skills_prompt, - false, ); if let Some(role_prompt) = resolved.role_prompt { system_prompt.push_str("\n\n## Agent Definition\n\n"); @@ -665,14 +669,6 @@ impl SubAgentManager { }) } - pub async fn run_parallel( - &self, - configs: Vec, - ) -> Result, SubAgentError> { - self.run_foreground_batch(configs, &ToolExecutionContext::default()) - .await - } - pub async fn run_foreground_batch( &self, configs: Vec, @@ -816,11 +812,6 @@ impl SubAgentManager { let tools = self.filter_tools(&config.allowed_tools); let timeout_secs = config.timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS); let timeout_human = format_duration(timeout_secs); - let http_get_only = config.allowed_tools.is_none() - || config - .allowed_tools - .as_ref() - .is_some_and(|v| v.iter().any(|t| t == "http_request")); let skills_prompt = self.get_skills_prompt(&tools); let system_prompt = build_sub_agent_system_prompt( &config.prompt, @@ -829,7 +820,6 @@ impl SubAgentManager { &self.provider_config.workspace_dir, &self.provider_config.model_id, skills_prompt, - http_get_only, ); let provider_config = self.provider_config.clone(); let storage = self.storage.clone(); diff --git a/src/agent/system_prompt.rs b/src/agent/system_prompt.rs index ac0b253..56ef7b4 100644 --- a/src/agent/system_prompt.rs +++ b/src/agent/system_prompt.rs @@ -57,7 +57,6 @@ impl SystemPromptBuilder { task: &str, timeout: &str, skills_prompt: Option, - http_get_only: bool, ) -> Self { let mut sections: Vec> = vec![ Box::new(SubAgentIdentitySection { @@ -66,7 +65,7 @@ impl SystemPromptBuilder { }), Box::new(ToolHonestySection), Box::new(SafetySection), - Box::new(SubAgentToolsSection { http_get_only }), + Box::new(SubAgentToolsSection), Box::new(WorkspaceSection), ]; if let Some(sp) = skills_prompt { @@ -390,9 +389,7 @@ impl PromptSection for SubAgentIdentitySection { } /// Sub-agent available tools description. -pub struct SubAgentToolsSection { - pub http_get_only: bool, -} +pub struct SubAgentToolsSection; impl PromptSection for SubAgentToolsSection { fn name(&self) -> &str { @@ -402,11 +399,6 @@ impl PromptSection for SubAgentToolsSection { fn build(&self, ctx: &PromptContext<'_>) -> String { let mut s = String::from("## 可用工具\n\n"); s.push_str(&ctx.tools.describe_for_prompt()); - if self.http_get_only { - s.push_str( - "\n\n**注意**：使用 http_request 时只允许 GET 方法，禁止 POST、PUT、DELETE 等。", - ); - } s } } @@ -514,15 +506,13 @@ pub fn build_sub_agent_system_prompt( workspace_dir: &Path, model_name: &str, skills_prompt: Option, - http_get_only: bool, ) -> String { let ctx = PromptContext { workspace_dir, model_name, tools, }; - SystemPromptBuilder::with_sub_agent_defaults(task, timeout_human, skills_prompt, http_get_only) - .build(&ctx) + SystemPromptBuilder::with_sub_agent_defaults(task, timeout_human, skills_prompt).build(&ctx) } #[cfg(test)] diff --git a/src/session/session.rs b/src/session/session.rs index 1995248..870d94e 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -1,7 +1,7 @@ use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex as StdMutex}; -use tokio::sync::{Mutex, mpsc, oneshot, watch}; +use tokio::sync::{Mutex, mpsc, watch}; use tokio_util::sync::CancellationToken; use super::persistence::{ @@ -485,7 +485,7 @@ mod cancelled_partial_tests { }) .unwrap(); controller.complete(None); - let (sender, completion) = oneshot::channel(); + let (sender, completion) = tokio::sync::oneshot::channel(); sender.send(Err(DeliveryError::FinalTimedOut)).unwrap(); let target = crate::channels::TurnTarget { @@ -624,12 +624,14 @@ pub struct Session { /// It is allocated while holding the Session mutex so worker cleanup can /// use that same lock as the send barrier. next_task_sequence: u64, - /// Cancel signal for the currently executing agent task - current_cancel: Option>, + /// Whether the session has an active Turn or queued local fallback + /// tasks. `/stop` clears it; the busy signal is observability-only + /// (idle wait, active-turn count). Cancellation itself is driven solely + /// by `current_turn_token`. + turn_busy: bool, /// Structured cancellation for the active Turn. `/stop` cancels it and /// the token propagates through AgentLoop provider streams and tool - /// batches. The oneshot above remains the busy/stop compatibility - /// marker until it is removed together with the legacy adapter. + /// batches. current_turn_token: Option, /// Latest durable inbox revision for this session. The worker watches /// this to claim agent events; the value only merges wakes, the payload @@ -661,6 +663,8 @@ 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 @@ -688,7 +692,11 @@ struct AgentTask { fn steer_input_from_event( event: &crate::storage::agent_inbox::AgentInboxEventRecord, now: i64, -) -> TurnInput { +) -> ( + TurnInput, + crate::agent::steering::WakeupSource, + Option, +) { use crate::agent::steering::InputDelivery; use crate::storage::agent_inbox::AgentEventType; let payload: serde_json::Value = @@ -699,13 +707,14 @@ fn steer_input_from_event( .and_then(serde_json::Value::as_str) .unwrap_or("unknown") .to_string(); - let (source, content) = match event.event_type { + let (source, wakeup_preview, content) = match event.event_type { AgentEventType::Signal => { let severity = event.severity.clone().unwrap_or_else(|| "info".to_string()); let summary = payload .get("summary") .and_then(serde_json::Value::as_str) - .unwrap_or_default(); + .unwrap_or_default() + .to_string(); let mut content = format!( "[后台 Agent 信号] severity={severity}, agent={agent_id}, run={run_id}\n{summary}" ); @@ -715,7 +724,14 @@ fn steer_input_from_event( content.push_str("\n详情: "); content.push_str(&details.to_string()); } - (TurnInputSource::AgentSignal { run_id, agent_id }, content) + ( + TurnInputSource::AgentSignal { + run_id: run_id.clone(), + agent_id: agent_id.clone(), + }, + (!summary.is_empty()).then_some(summary), + content, + ) } AgentEventType::Completion => { let status = payload @@ -728,17 +744,27 @@ fn steer_input_from_event( content.push_str(&format!("\n错误: {error}")); } ( - TurnInputSource::AgentCompletion { run_id, agent_id }, + TurnInputSource::AgentCompletion { + run_id: run_id.clone(), + agent_id: agent_id.clone(), + }, + None, content, ) } AgentEventType::GroupCompletion => { let group_id = event.group_id.clone().unwrap_or_default(); let content = format!("[后台 Agent 任务组结果] group={group_id}"); - (TurnInputSource::AgentGroupCompletion { group_id }, content) + ( + TurnInputSource::AgentGroupCompletion { + group_id: group_id.clone(), + }, + None, + content, + ) } }; - TurnInput { + let input = TurnInput { id: format!("steer:{}", event.id), sequence: 0, source, @@ -749,7 +775,9 @@ 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 @@ -918,7 +946,7 @@ impl Session { memory_manager, agent_tx: None, next_task_sequence: 1, - current_cancel: None, + turn_busy: false, current_turn_token: None, agent_inbox_wake: watch::channel(0).0, active_turn_emitter: None, @@ -1115,7 +1143,7 @@ impl Session { memory_manager, agent_tx: None, next_task_sequence: 1, - current_cancel: None, + turn_busy: false, current_turn_token: None, agent_inbox_wake: watch::channel(0).0, active_turn_emitter: None, @@ -1280,6 +1308,7 @@ 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())), }); } @@ -1866,6 +1895,17 @@ fn resolve_slash_command(command: &str) -> Option<&'static SlashCommand> { }) } +/// Result of one steer admission pass over the active Turn. +enum SteerAdmission { + /// Events were activated into the active Turn's mailbox. + Activated, + /// Events were claimed but could not be admitted and returned to + /// the queue lane (or the active Turn disappeared). + ReleasedToQueue, + /// No steer events were due. + NothingClaimed, +} + impl SessionManager { fn worker_deps(&self) -> AgentWorkerDeps { AgentWorkerDeps { @@ -1947,7 +1987,7 @@ impl SessionManager { Some(skills_loader.clone()), task_supervisor.clone(), ) - .with_admission(admission) + .with_admission(admission.clone()) .with_catalog(agent_catalog.clone()) .with_execution_gate(execution_gate.clone()) .with_work_manager(work_manager.clone()), @@ -1962,6 +2002,8 @@ impl SessionManager { 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, @@ -2368,7 +2410,8 @@ impl SessionManager { let msgs = { let mut guard = session.lock().await; let mut msgs: Vec = Vec::new(); - if guard.current_cancel.take().is_some() { + if guard.turn_busy { + guard.turn_busy = false; msgs.push("当前任务已发送停止信号。".to_string()); } if let Some(token) = guard.current_turn_token.take() { @@ -2501,7 +2544,7 @@ impl SessionManager { .agent_tx .as_ref() .is_some_and(|sender| sender.capacity() < sender.max_capacity()); - if session.current_cancel.is_some() || queued { + if session.turn_busy || queued { busy = true; break; } @@ -2535,7 +2578,7 @@ impl SessionManager { let mut count = 0; for session in sessions { let session = session.lock().await; - if session.current_cancel.is_some() { + if session.turn_busy { count += 1; } } @@ -3306,7 +3349,24 @@ impl SessionManager { message.timestamp, ); match active.steering.try_push_user(input) { - Ok(()) => return Ok(HandleResult::AgentProcessing), + 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(_) => { active .recovery @@ -3350,7 +3410,7 @@ impl SessionManager { guard.agent_tx.is_none() || guard.agent_tx.as_ref().is_some_and(|tx| tx.is_closed()); if needs_spawn { guard.agent_tx = None; - guard.current_cancel = None; + guard.turn_busy = false; guard.current_turn_token = None; guard.worker_generation = guard.worker_generation.wrapping_add(1); let generation = guard.worker_generation; @@ -3381,7 +3441,7 @@ impl SessionManager { // worker under the same lock and retry the recovered task once. let task = error.into_inner(); guard.agent_tx = None; - guard.current_cancel = None; + guard.turn_busy = false; guard.current_turn_token = None; guard.worker_generation = guard.worker_generation.wrapping_add(1); let generation = guard.worker_generation; @@ -3405,6 +3465,24 @@ 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) } } @@ -3692,7 +3770,6 @@ fn spawn_agent_worker( mut compressor, system_prompt_out, base_version, - cancel_rx, turn_token, turn_controller, turn_emitter, @@ -3728,13 +3805,12 @@ fn spawn_agent_worker( } }; - let (cancel_tx, cancel_rx) = oneshot::channel(); let turn_token = CancellationToken::new(); if guard.worker_generation != worker_gen { return; // /stop replaced us } - guard.current_cancel = Some(cancel_tx); + guard.turn_busy = true; guard.current_turn_token = Some(turn_token.clone()); // Install the active-turn handle before memory recall and @@ -3750,10 +3826,12 @@ 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(), }); @@ -3763,8 +3841,7 @@ fn spawn_agent_worker( guard.fresh_context_compressor(), guard.build_system_prompt(&skills_prompt), guard.state_version, - cancel_rx, - turn_token, + turn_token.clone(), turn_controller, turn_emitter, turn_receiver, @@ -3826,7 +3903,7 @@ fn spawn_agent_worker( session_id = %guard.id, "Session changed while preparing agent history; dropping stale task" ); - guard.current_cancel = None; + guard.turn_busy = false; guard.current_turn_token = None; let mut released = Vec::new(); if guard @@ -3922,7 +3999,7 @@ fn spawn_agent_worker( turn_controller.cancel(Some( "session changed before model execution".to_string(), )); - guard.current_cancel = None; + guard.turn_busy = false; guard.current_turn_token = None; drop(guard); if let Some(storage) = storage { @@ -3953,13 +4030,24 @@ 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 tool_context = ToolExecutionContext::for_session(&response_session_id) + let mut tool_context = ToolExecutionContext::for_session(&response_session_id) .with_turn_id(agent_turn.turn_id.clone()) - .with_cancellation(turn_token.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 = crate::agent::sub_agent::DELEGATE_CONTEXT.scope( crate::agent::DelegateContext { session_id: unified_str2, @@ -4341,8 +4429,9 @@ fn spawn_agent_worker( tokio::select! { () = process_future => {} - _ = cancel_rx => { - // cancelled — current_cancel already taken by /stop + _ = turn_token.cancelled() => { + // Cancelled by `/stop`, which took and cancelled the + // token; the terminal state is persisted here. persist_cancelled_turn( &turn_controller, &session, @@ -4403,16 +4492,11 @@ fn spawn_agent_worker( if guard.worker_generation == worker_gen { guard.current_turn_token = None; consecutive_user_turns = consecutive_user_turns.saturating_add(1); - if local_tasks.is_empty() { - guard.current_cancel = None; - } else { - // Keep the session observable as busy while local - // fallback tasks are waiting for their next Turn. - // `/stop` can still invalidate this generation before - // the worker starts the next task. - let (cancel_tx, _cancel_rx) = oneshot::channel(); - guard.current_cancel = Some(cancel_tx); - } + // Keep the session observable as busy while local + // fallback tasks are waiting for their next Turn. + // `/stop` can still invalidate this generation before + // the worker starts the next task. + guard.turn_busy = !local_tasks.is_empty(); } } }).await; @@ -4692,8 +4776,25 @@ impl crate::agent::AgentInboxWakeTarget for SessionManager { // admission; everything that cannot be admitted stays pending // for the queue lane. if has_active_turn { - self.try_steer_inbox_events(&session, session_id, revision) + let outcome = 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; @@ -4732,9 +4833,11 @@ impl SessionManager { session: &Arc>, session_id: &str, revision: i64, - ) { + ) -> SteerAdmission { let storage = { session.lock().await.storage.clone() }; - let Some(storage) = storage else { return }; + let Some(storage) = storage else { + return SteerAdmission::NothingClaimed; + }; let now = chrono::Utc::now().timestamp_millis(); let Ok(Some(lease)) = crate::storage::Storage::claim_inbox_batch( &storage, @@ -4747,7 +4850,7 @@ impl SessionManager { ) .await else { - return; + return SteerAdmission::NothingClaimed; }; let token = lease.token.clone(); @@ -4765,16 +4868,20 @@ impl SessionManager { .collect(); release_steer_leases(&storage, leases).await; let _ = session.lock().await.agent_inbox_wake.send_replace(revision); - return; + return SteerAdmission::ReleasedToQueue; }; (active.turn_id.clone(), active.steering.clone()) }; let mut rejected = Vec::new(); let mut reserved = Vec::new(); + let mut wakeup_sources = Vec::new(); for event in &lease.events { - let input = steer_input_from_event(event, now); + let (input, wakeup_source, preview) = steer_input_from_event(event, now); match mailbox.try_reserve_steer(input, token.clone()) { - Ok(()) => reserved.push(event.id.clone()), + Ok(()) => { + reserved.push(event.id.clone()); + wakeup_sources.push((wakeup_source, preview)); + } Err(_) => rejected.push((event.id.clone(), token.clone())), } } @@ -4806,13 +4913,33 @@ impl SessionManager { }; if still_active { mailbox.activate_reserved(); - } else { - rejected.extend(admitted); - release_steer_leases(&storage, rejected).await; - let _ = session.lock().await.agent_inbox_wake.send_replace(revision); + // 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); + release_steer_leases(&storage, rejected).await; + let _ = session.lock().await.agent_inbox_wake.send_replace(revision); + SteerAdmission::ReleasedToQueue } else { release_steer_leases(&storage, rejected).await; + SteerAdmission::ReleasedToQueue } } } @@ -4975,7 +5102,7 @@ fn format_task_notification( #[cfg(test)] mod slash_command_tests { use super::{ - AgentTask, SLASH_COMMANDS, pop_lowest_sequence, prepend_pending_steering, + AgentTask, SLASH_COMMANDS, Session, pop_lowest_sequence, prepend_pending_steering, resolve_slash_command, }; use crate::agent::steering::{SteeringPushError, TurnInput, TurnMailbox}; @@ -5161,4 +5288,92 @@ 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/storage/agent_inbox.rs b/src/storage/agent_inbox.rs index 9337e41..6b8cb65 100644 --- a/src/storage/agent_inbox.rs +++ b/src/storage/agent_inbox.rs @@ -1150,6 +1150,7 @@ pub(crate) async fn insert_completion_event_tx( session_id: &str, status: &str, error: Option<&str>, + signal_ids: &[String], now: i64, ) -> Result<(), StorageError> { let Some(revision) = convert_reservation_tx(tx, session_id, now).await? else { @@ -1176,6 +1177,7 @@ pub(crate) async fn insert_completion_event_tx( payload_json: serde_json::json!({ "status": status, "error": error, + "signal_ids": signal_ids, }) .to_string(), }; @@ -1270,6 +1272,8 @@ mod tests { task: "work".to_string(), context_json: None, budget_json: "{}".to_string(), + signal_contract_json: None, + signal_delivery: None, deadline_at: 1000, runtime_generation: 1, completion_slot_reserved: false, @@ -1685,6 +1689,8 @@ mod tests { task: "work".to_string(), context_json: None, budget_json: "{}".to_string(), + signal_contract_json: None, + signal_delivery: None, deadline_at: 1000, runtime_generation: 1, completion_slot_reserved: false, diff --git a/src/storage/agent_run.rs b/src/storage/agent_run.rs index 58d088b..efce074 100644 --- a/src/storage/agent_run.rs +++ b/src/storage/agent_run.rs @@ -395,6 +395,8 @@ pub struct NewAgentRun { pub task: String, pub context_json: Option, pub budget_json: String, + pub signal_contract_json: Option, + pub signal_delivery: Option, pub deadline_at: i64, pub runtime_generation: i64, /// Background runs reserve a completion slot at admission so their @@ -448,21 +450,28 @@ pub enum AgentTerminalOutcome { cost: Option, tool_calls: i64, iterations: i64, + /// Signal IDs emitted by this run; included in the completion + /// payload so the main Agent can recognise duplicates (design §12.3). + signal_ids: Vec, }, Failed { error: String, prompt_tokens: Option, completion_tokens: Option, cost: Option, + signal_ids: Vec, }, TimedOut { deadline_at: i64, + signal_ids: Vec, }, Cancelled { reason: String, + signal_ids: Vec, }, Interrupted { reason: String, + signal_ids: Vec, }, } @@ -628,10 +637,11 @@ impl super::Storage { parent_run_id, caller_agent_id, caller_scope_id, idempotency_key, \ agent_id, definition_hash, provider_profile, provider_name, model_id, \ mode, depth, plan_item_id, execution_id, task, context_json, budget_json, \ + signal_contract_json, signal_delivery, \ status, runtime_generation, attempt, completion_slot_reserved, deadline_at, \ revision, created_at, updated_at) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, \ - 'queued', ?, 1, ?, ?, 0, ?, ?)", + ?, ?, 'queued', ?, 1, ?, ?, 0, ?, ?)", ) .bind(&run.id) .bind(request.group.as_ref().map(|group| group.id.clone())) @@ -653,6 +663,8 @@ impl super::Storage { .bind(&run.task) .bind(&run.context_json) .bind(&run.budget_json) + .bind(&run.signal_contract_json) + .bind(&run.signal_delivery) .bind(run.runtime_generation) .bind(i64::from(run.completion_slot_reserved)) .bind(run.deadline_at) @@ -1052,6 +1064,7 @@ impl super::Storage { cost, tool_calls, iterations, + .. } => ( Some(result.as_str()), None, @@ -1066,6 +1079,7 @@ impl super::Storage { prompt_tokens, completion_tokens, cost, + .. } => ( None, Some(error.as_str()), @@ -1078,10 +1092,10 @@ impl super::Storage { AgentTerminalOutcome::TimedOut { .. } => { (None, Some("deadline exceeded"), None, None, None, 0, 0) } - AgentTerminalOutcome::Cancelled { reason } => { + AgentTerminalOutcome::Cancelled { reason, .. } => { (None, Some(reason.as_str()), None, None, None, 0, 0) } - AgentTerminalOutcome::Interrupted { reason } => { + AgentTerminalOutcome::Interrupted { reason, .. } => { (None, Some(reason.as_str()), None, None, None, 0, 0) } }; @@ -1182,13 +1196,23 @@ impl super::Storage { // reservation into a durable completion event in the same commit. // The event survives restarts, queue-full conditions and lost wakes. if run.completion_slot_reserved { - let (status, error) = match outcome { - AgentTerminalOutcome::Completed { .. } => ("completed", None), - AgentTerminalOutcome::Failed { error, .. } => ("failed", Some(error.as_str())), - AgentTerminalOutcome::TimedOut { .. } => ("timed_out", Some("deadline exceeded")), - AgentTerminalOutcome::Cancelled { reason } => ("cancelled", Some(reason.as_str())), - AgentTerminalOutcome::Interrupted { reason } => { - ("interrupted", Some(reason.as_str())) + let (status, error, signal_ids) = match outcome { + AgentTerminalOutcome::Completed { signal_ids, .. } => { + ("completed", None, signal_ids.as_slice()) + } + AgentTerminalOutcome::Failed { + error, signal_ids, .. + } => ("failed", Some(error.as_str()), signal_ids.as_slice()), + AgentTerminalOutcome::TimedOut { signal_ids, .. } => ( + "timed_out", + Some("deadline exceeded"), + signal_ids.as_slice(), + ), + AgentTerminalOutcome::Cancelled { reason, signal_ids } => { + ("cancelled", Some(reason.as_str()), signal_ids.as_slice()) + } + AgentTerminalOutcome::Interrupted { reason, signal_ids } => { + ("interrupted", Some(reason.as_str()), signal_ids.as_slice()) } }; super::agent_inbox::insert_completion_event_tx( @@ -1197,6 +1221,7 @@ impl super::Storage { &run.root_session_id, status, error, + signal_ids, now, ) .await?; @@ -1335,6 +1360,8 @@ mod tests { task: "do the work".to_string(), context_json: None, budget_json: "{\"remaining_runs\":15}".to_string(), + signal_contract_json: None, + signal_delivery: None, deadline_at: 1_000, runtime_generation: 1, completion_slot_reserved: false, @@ -1429,6 +1456,7 @@ mod tests { cost: None, tool_calls: 1, iterations: 2, + signal_ids: Vec::new(), }, None, 120, @@ -1449,6 +1477,7 @@ mod tests { prompt_tokens: None, completion_tokens: None, cost: None, + signal_ids: Vec::new(), }, None, 130, @@ -1489,6 +1518,7 @@ mod tests { cost: None, tool_calls: 0, iterations: 0, + signal_ids: Vec::new(), }, None, 120, @@ -1602,6 +1632,7 @@ mod tests { cost: None, tool_calls: 0, iterations: 0, + signal_ids: Vec::new(), }, Some("finished the work"), 120, diff --git a/src/tools/delegate.rs b/src/tools/delegate.rs index adcc3e5..71fa2db 100644 --- a/src/tools/delegate.rs +++ b/src/tools/delegate.rs @@ -240,10 +240,25 @@ impl DelegateTool { "mixed named and legacy general batches are not supported", )); } - self.sub_agent_manager + let mut results = self + .sub_agent_manager .run_foreground_batch(configs, context) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))? + .map_err(|error| anyhow::anyhow!(error.to_string()))?; + // Legacy general compatibility: tell the model the + // unnamed path is deprecated so it migrates to named + // Agents (which get durable runs, fixed tool sets and + // per-run resource scopes). + for result in results.iter_mut() { + if matches!(result.status, TaskStatus::Completed) { + result.content = format!( + "{}\n\n[提示] 无 target 的通用 Agent 是兼容模式：不持久化、不可审计。\ + 建议为固定角色创建具名 Agent definition 并使用 target 委托。", + result.content + ); + } + } + results }; let payload: Vec<_> = results .into_iter() diff --git a/src/tools/emit_signal.rs b/src/tools/emit_signal.rs index 0cac9d6..7956d00 100644 --- a/src/tools/emit_signal.rs +++ b/src/tools/emit_signal.rs @@ -254,8 +254,6 @@ impl Tool for EmitSignalTool { if let Ok(mut signals) = agent.emitted_signals.lock() { signals.push(EmittedSignal { signal_id: accepted.signal_id.clone(), - severity: args.severity.clone(), - summary: args.summary.clone(), }); } } diff --git a/src/tools/sleep.rs b/src/tools/sleep.rs index 789bef9..36b4caa 100644 --- a/src/tools/sleep.rs +++ b/src/tools/sleep.rs @@ -3,6 +3,8 @@ 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; @@ -88,27 +90,132 @@ impl Tool for SleepTool { } }; - tokio::select! { - biased; - _ = context.cancellation.cancelled() => { - anyhow::bail!("sleep cancelled"); + 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()); } - _ = tokio::time::sleep(Duration::from_secs(seconds)) => {} } + 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: format!("Slept for {seconds} second(s)."), + 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::AgentGroupCompletion { group_id }) => { + message.push_str(&format!( + " 收到一条 steer AgentGroupCompletion（group={group_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; @@ -135,6 +242,10 @@ mod tests { assert!(!tool.read_only()); assert!(!tool.concurrency_safe()); assert!(!tool.exclusive()); + assert_eq!( + tool.input_interrupt_policy(), + crate::tools::InputInterruptPolicy::WakeOnly + ); } #[tokio::test] @@ -289,4 +400,130 @@ mod tests { .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 9c3e7a1..74f01cf 100644 --- a/src/tools/traits.rs +++ b/src/tools/traits.rs @@ -10,6 +10,10 @@ 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 { @@ -20,6 +24,7 @@ impl Default for ToolExecutionContext { agent: None, cancellation: tokio_util::sync::CancellationToken::new(), execution_gate: None, + turn_wakeup: None, } } } @@ -32,6 +37,7 @@ impl ToolExecutionContext { agent: None, cancellation: tokio_util::sync::CancellationToken::new(), execution_gate: None, + turn_wakeup: None, } } @@ -60,6 +66,11 @@ 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)] diff --git a/webui/package-lock.json b/webui/package-lock.json index 5a245bd..7752d35 100644 --- a/webui/package-lock.json +++ b/webui/package-lock.json @@ -1,12 +1,12 @@ { "name": "picobot-webui", - "version": "1.7.0", + "version": "1.8.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "picobot-webui", - "version": "1.7.0", + "version": "1.8.0", "dependencies": { "bits-ui": "^2.0.0", "dompurify": "^3.4.12", diff --git a/webui/package.json b/webui/package.json index 095b7ed..58451a6 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,7 +1,7 @@ { "name": "picobot-webui", "private": true, - "version": "1.7.0", + "version": "1.8.0", "type": "module", "engines": { "node": ">=20"