diff --git a/AGENTS.md b/AGENTS.md index 1fc12a0..3db28bc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,7 +47,7 @@ Channel → MessageBus.inbound → Gateway processor → SessionManager → per- └── Channel ← OutboundDispatcher ← per-conversation lane ← MessageBus.outbound WebSocket/Channel → MessageBus.control → Gateway processor → SessionManager (dialog operations) -Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message tool +Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler delivery policy → SessionManager/MessageBus ``` ### Modules @@ -76,6 +76,7 @@ Scheduler → SessionManager.handle_cron_message → AgentLoop → send_message - **Channels** only send/receive messages via `MessageBus`; they know nothing about sessions or LLM - **MessageBus** owns bounded inbound/outbound/control queues; outbound routing and retries belong to `OutboundDispatcher`, not the queue - **SessionManager** owns session state, dialog operations, context construction, per-session work queues, and persistence coordination +- **Scheduler** supports legacy direct-delivery jobs and managed `task`/`monitor` jobs; managed agents cannot call `send_message`, and `on_alert` suppresses only healthy informational results - **AgentLoop** is stateless across turns; it receives prepared history, calls LLM providers, executes tools, and returns one result - **WebUI management APIs** only expose allowlisted config/profile/log/storage operations; keep response limits, secret redaction, atomic config writes, and profile path allowlists intact - **WebUI slash completion** must consume the existing `get_slash_commands` WebSocket response; do not duplicate the backend command list in frontend source diff --git a/README.md b/README.md index 2e2742c..8bbb28b 100644 --- a/README.md +++ b/README.md @@ -179,7 +179,7 @@ TUI 会把一个随机客户端标识保存到 `~/.picobot/tui_client_id`，因 | `providers` | OpenAI 兼容接口和 Anthropic Messages API 客户端 | | `tools` | Agent 可调用工具集合 | | `storage` | SQLite schema、CRUD、消息和任务持久化 | -| `scheduler` | 轮询 Cron 任务并把任务 prompt 送入目标会话 | +| `scheduler` | 领取定时任务，执行普通/巡检 Agent，并按投递策略记录或发送结果 | | `skills` | 加载 Skill，并把 Skill 指南注入系统提示 | | `mcp` | 连接 MCP Server，将远端工具包装成普通 Tool | | `task_supervisor` | 统一管理 Gateway 后台任务的取消和有界关停 | @@ -228,7 +228,7 @@ PicoBot 有两类记忆： | Knowledge | 偏好、事实、项目规则、长期可复用信息 | 长期保留，手动删除 | | Timeline | 长对话压缩后的历史摘要 | 默认保留 90 天 | -每轮处理用户消息时，MemoryManager 会按用户输入召回 Knowledge，并作为运行时上下文附加到本轮用户消息。当前召回上限固定为 5；`memory.recall_limit` 已支持解析但尚未接入 worker。上下文压缩产生的摘要会保存为 Timeline，后续可通过 `timeline_recall` 工具检索。 +每轮处理用户消息时，MemoryManager 会按用户输入召回 Knowledge，并作为运行时上下文附加到本轮用户消息。当前召回上限固定为 5；`memory.recall_limit` 已支持解析但尚未接入 worker。上下文压缩产生的摘要会保存为 Timeline，后续可通过 `timeline_recall` 工具检索。Scheduler 默认创建一个每日维护巡检，按 `memory.timeline_retention_days` 清理过期 Timeline；Knowledge 不会被自动删除。 ### 工具 @@ -247,6 +247,7 @@ PicoBot 有两类记忆： | `send_message` | 向指定渠道发送消息 | | `chat_manager` | 查看渠道、会话和历史消息 | | `cron_add/list/remove/enable/disable/update` | 管理定时任务 | +| `routine_maintenance` | 安全清理超过保留期的 Timeline，不删除 Knowledge | | `browser` | 可选 WebDriver 浏览器自动化 | | MCP tools | 从配置的 MCP Server 动态发现并注册 | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 417a4f6..e5027fe 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -64,7 +64,7 @@ flowchart LR | `tools` / `mcp` | 工具定义、注册和执行适配 | 隐式修改会话路由 | | `storage` | SQLite schema、迁移、原子 CRUD | 运行时调度策略 | | `memory` | Knowledge/Timeline 的存取与召回 | 直接驱动消息发送 | -| `scheduler` | 领取到期任务、限并发执行、原子记录结果 | 复用聊天会话历史 | +| `scheduler` | 领取到期任务、执行普通/巡检 Agent、应用投递策略、原子记录结果 | 复用聊天会话历史、直接感知 Channel | | `task_supervisor` | 后台任务注册、取消、限时回收 | 业务级重试和结果语义 | ## 4. 消息与控制数据流 @@ -212,7 +212,7 @@ WebUI 与 Gateway 当前属于同一信任边界，没有内置认证。默认 1. 加载配置和 `.env`，解析 workspace。 2. 创建并切换到 workspace。 -3. 初始化 SQLite、MemoryManager、MessageBus 和 SessionManager。 +3. 初始化 SQLite、MemoryManager、MessageBus 和 SessionManager；Scheduler 启用时幂等创建默认日常维护巡检。 4. 注册内置工具、渠道、MCP 工具和 Cron 工具。 5. 启动所有 Channel。 6. 通过 TaskSupervisor 启动 message processor、dispatcher 和 scheduler。 diff --git a/resources/skills/about-picobot/references/config.md b/resources/skills/about-picobot/references/config.md index 41b185d..2a8b4d5 100644 --- a/resources/skills/about-picobot/references/config.md +++ b/resources/skills/about-picobot/references/config.md @@ -70,7 +70,7 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `enabled` | bool | true | 是否启动调度器并注册 cron 工具 | | `poll_interval_secs` | int | 60 | 检查到期任务的轮询间隔 | | `max_concurrent` | int | 1 | 每批到期任务的最大并发数，运行时限制在 1–256 | -| `execution_timeout_secs` | int | 900 | 单个定时任务 Agent 执行的硬超时；租约会额外增加 30 秒 | +| `execution_timeout_secs` | int | 900 | 单个定时任务 Agent 执行的硬超时；租约会覆盖执行和托管投递等待 | ## memory 字段 @@ -80,10 +80,10 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `consolidation_model` | string | 主 Agent model | 当前记录在 MemoryManager 中供后续归并使用；压缩摘要仍使用 Session model | | `recall_limit` | int | 5 | 预期的每轮知识召回上限；当前 worker 固定使用 5 | | `idle_consolidation_minutes` | int | 10 | 预留的空闲归并阈值；当前无对应循环 | -| `timeline_retention_days` | int | 90 | 预留的 Timeline 保留期；当前无自动清理循环 | +| `timeline_retention_days` | int | 90 | 默认日常维护巡检删除超过该期限的 Timeline；Knowledge 不受影响 | | `max_failures_before_degrade` | int | 3 | 预留的归并失败阈值；当前无失败降级循环 | -注意：这些字段都会被解析，但当前 worker 的 Knowledge 召回数量仍固定为 5；idle consolidation、Timeline 自动清理和失败降级循环尚未接入。配置存在不等于对应后台行为已经生效。 +注意：当前 worker 的 Knowledge 召回数量仍固定为 5；idle consolidation 和失败降级循环尚未接入。Timeline 清理由默认启用的 `picobot-routine-maintenance` 定时巡检执行。 ## channels.feishu 字段 diff --git a/resources/skills/about-picobot/references/tools.md b/resources/skills/about-picobot/references/tools.md index 2569943..00e28ff 100644 --- a/resources/skills/about-picobot/references/tools.md +++ b/resources/skills/about-picobot/references/tools.md @@ -67,7 +67,7 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 {"type":"cron","expr":"0 0 9 * * *","tz":"Asia/Shanghai"} ``` -时间戳和间隔单位为毫秒；Cron 表达式为 6 段（秒、分、时、日、月、周）。定时 Agent 不复用聊天历史，`prompt` 必须包含完整上下文，并由 Agent 使用 `send_message` 投递结果。`model` 当前会持久化和展示，但执行仍使用默认 Agent Provider/Model，不能依赖它实现模型覆盖。 +时间戳和间隔单位为毫秒；Cron 表达式为 6 段（秒、分、时、日、月、周）。定时 Agent 不复用聊天历史，`prompt` 必须包含完整上下文。`kind` 可为 `task` 或 `monitor`；`delivery_policy` 可为 `always`、`on_alert` 或 `never`。托管任务由 Scheduler 投递，巡检返回 `NO_REPLY[INFO]` 时静默，`NO_REPLY[FAIL]`/`NO_REPLY[REFUSE]` 仍视为需关注结果。升级前创建的任务保留 Agent 直接调用 `send_message` 的兼容行为。`model` 当前会持久化和展示，但执行仍使用默认 Agent Provider/Model，不能依赖它实现模型覆盖。 --- @@ -112,6 +112,10 @@ Cron 不是一个带 `action` 的统一工具，而是六个独立工具；仅 |------|------|------| | `key` | 是 | 要删除的记忆键 | +## routine_maintenance — 日常维护 + +清理超过 `memory.timeline_retention_days` 的 Timeline 记忆；不会删除 Knowledge。Gateway 在 Scheduler 启用时幂等创建一个每日运行的默认维护巡检，用户禁用或修改后不会在重启时被覆盖。 + --- ## get_skill — 获取 Skill diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 5c259c9..7ccf3d0 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -148,6 +148,16 @@ impl GatewayState { // Initialize scheduler if enabled in config let scheduler_config = config.gateway.scheduler.clone().unwrap_or_default(); if scheduler_config.enabled { + session_manager + .tools() + .register(crate::tools::RoutineMaintenanceTool::new( + storage.clone(), + config.memory.timeline_retention_days, + )); + storage + .ensure_default_maintenance_job() + .await + .map_err(|e| format!("failed to seed default maintenance job: {e}"))?; // Register cron tools session_manager .tools() diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index 14e768f..22f76df 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -9,12 +9,51 @@ use tokio::time; use crate::config::SchedulerConfig; use crate::session::SessionManager; use crate::session::session::HandleResult; -use crate::storage::JobRun; use crate::storage::ScheduledJob; use crate::storage::Storage; +use crate::storage::{DeliveryPolicy, JobKind, JobRun}; pub use types::Schedule; +#[derive(Debug, Clone, PartialEq, Eq)] +enum ScheduledDisposition { + Content(String), + Quiet(String), + ReportedFailure(String), + Refused(String), +} + +fn parse_scheduled_disposition(output: &str) -> ScheduledDisposition { + let trimmed = output.trim(); + if trimmed.eq_ignore_ascii_case("NO_REPLY") { + return ScheduledDisposition::Quiet(String::new()); + } + let upper = trimmed.to_ascii_uppercase(); + for (prefix, kind) in [ + ("NO_REPLY[INFO]", "info"), + ("NO_REPLY[FAIL]", "fail"), + ("NO_REPLY[REFUSE]", "refuse"), + ] { + if upper.starts_with(prefix) { + let suffix = &trimmed[prefix.len()..]; + if !suffix.is_empty() && !suffix.trim_start().starts_with(':') { + continue; + } + let reason = suffix.trim().trim_start_matches(':').trim().to_string(); + return match kind { + "info" => ScheduledDisposition::Quiet(reason), + "fail" => ScheduledDisposition::ReportedFailure(reason), + _ => ScheduledDisposition::Refused(reason), + }; + } + } + if trimmed.is_empty() { + ScheduledDisposition::ReportedFailure("scheduled agent returned empty output".into()) + } else { + ScheduledDisposition::Content(trimmed.to_string()) + } +} + /// Compute the next execution time (Unix ms) for a schedule, given `from` (Unix ms). /// Returns `None` if no next time can be determined (e.g., invalid cron expression). pub fn next_run_for_schedule(schedule: &Schedule, from: i64) -> Option { @@ -97,7 +136,7 @@ impl Scheduler { let lease_ms = self .config .execution_timeout_secs - .saturating_add(30) + .saturating_add(150) .saturating_mul(1000) .min(i64::MAX as u64) as i64; let lease_until = now.saturating_add(lease_ms); @@ -132,13 +171,30 @@ impl Scheduler { let started_at = now_ms(); tracing::info!(job_id = %job.id, job_name = %job.name, "scheduler: executing claimed job"); - let execution = self.session_manager.handle_cron_message( - &job.channel, - &job.chat_id, - &job.prompt, - &job.id, - &job.name, - ); + let managed = job.delivery_policy != DeliveryPolicy::Direct; + let execution = async { + if managed { + self.session_manager + .handle_managed_scheduled_message( + &job.prompt, + &job.id, + &job.name, + job.job_kind == JobKind::Monitor, + ) + .await + .map(HandleResult::AgentResponse) + } else { + self.session_manager + .handle_cron_message( + &job.channel, + &job.chat_id, + &job.prompt, + &job.id, + &job.name, + ) + .await + } + }; let result = time::timeout( time::Duration::from_secs(self.config.execution_timeout_secs.max(1)), execution, @@ -147,33 +203,143 @@ impl Scheduler { let finished_at = now_ms(); let duration_ms = start.elapsed().as_millis() as i64; - let (status, output, error) = match result { - Ok(Ok(HandleResult::AgentResponse(output) | HandleResult::CommandOutput(output))) => { - let output = if output.len() > 8000 { - format!( - "{}...[truncated]", - &output[..output.ceil_char_boundary(8000)] - ) - } else { - output - }; - ("ok".to_string(), Some(output), None) + let (mut status, output, error, result_kind, mut delivery_status, mut delivery_error) = + match result { + Ok(Ok( + HandleResult::AgentResponse(output) | HandleResult::CommandOutput(output), + )) => { + let output = if output.len() > 8000 { + format!( + "{}...[truncated]", + &output[..output.ceil_char_boundary(8000)] + ) + } else { + output + }; + if !managed { + ( + "ok".into(), + Some(output), + None, + None, + Some("direct".into()), + None, + ) + } else { + let disposition = parse_scheduled_disposition(&output); + let (kind, content, alert) = match &disposition { + ScheduledDisposition::Content(value) => { + ("content", Some(value.as_str()), true) + } + ScheduledDisposition::Quiet(_) => ("quiet", None, false), + ScheduledDisposition::ReportedFailure(value) => { + ("reported_failure", Some(value.as_str()), true) + } + ScheduledDisposition::Refused(value) => { + ("refused", Some(value.as_str()), true) + } + }; + let should_deliver = match job.delivery_policy { + DeliveryPolicy::Always => true, + DeliveryPolicy::OnAlert => alert, + DeliveryPolicy::Never => false, + DeliveryPolicy::Direct => false, + }; + if should_deliver { + let message = content.unwrap_or("巡检完成，未发现需要关注的问题。"); + match self + .session_manager + .deliver_scheduled_message( + &job.channel, + &job.chat_id, + &job.id, + &job.name, + message, + ) + .await + { + Ok(()) => ( + "ok".into(), + Some(output), + None, + Some(kind.into()), + Some("delivered".into()), + None, + ), + Err(delivery_error) => ( + "delivery_error".into(), + Some(output), + None, + Some(kind.into()), + Some("failed".into()), + Some(delivery_error), + ), + } + } else { + let delivery = if job.delivery_policy == DeliveryPolicy::Never { + "skipped" + } else { + "suppressed" + }; + ( + "ok".into(), + Some(output), + None, + Some(kind.into()), + Some(delivery.into()), + None, + ) + } + } + } + Ok(Ok(HandleResult::AgentProcessing)) => ( + "error".to_string(), + None, + Some("cron execution returned asynchronous processing".to_string()), + None, + None, + None, + ), + Ok(Err(error)) => ( + "error".to_string(), + None, + Some(error.to_string()), + None, + None, + None, + ), + Err(_) => ( + "timeout".to_string(), + None, + Some(format!( + "execution exceeded {} seconds", + self.config.execution_timeout_secs.max(1) + )), + None, + None, + None, + ), + }; + + if managed + && delivery_status.is_none() + && job.delivery_policy != DeliveryPolicy::Never + && let Some(message) = error.as_deref() + { + let notice = format!("定时任务「{}」执行失败：{}", job.name, message); + match self + .session_manager + .deliver_scheduled_message(&job.channel, &job.chat_id, &job.id, &job.name, ¬ice) + .await + { + Ok(()) => delivery_status = Some("delivered".into()), + Err(error) => { + status = "delivery_error".into(); + delivery_status = Some("failed".into()); + delivery_error = Some(error); + } } - Ok(Ok(HandleResult::AgentProcessing)) => ( - "error".to_string(), - None, - Some("cron execution returned asynchronous processing".to_string()), - ), - Ok(Err(error)) => ("error".to_string(), None, Some(error.to_string())), - Err(_) => ( - "timeout".to_string(), - None, - Some(format!( - "execution exceeded {} seconds", - self.config.execution_timeout_secs.max(1) - )), - ), - }; + } let (next_run_at, disable, delete) = match &job.schedule { Schedule::At { .. } => (None, !job.delete_after_run, job.delete_after_run), @@ -193,6 +359,9 @@ impl Scheduler { output, error, duration_ms, + result_kind, + delivery_status, + delivery_error, }; if let Err(error) = self @@ -271,6 +440,38 @@ mod tests { assert_eq!(next, expected); } + #[test] + fn scheduled_disposition_is_fail_safe() { + assert!(matches!( + parse_scheduled_disposition("NO_REPLY"), + ScheduledDisposition::Quiet(_) + )); + assert!(matches!( + parse_scheduled_disposition("NO_REPLY[INFO]: healthy"), + ScheduledDisposition::Quiet(_) + )); + assert!(matches!( + parse_scheduled_disposition("NO_REPLY[FAIL]: timeout"), + ScheduledDisposition::ReportedFailure(_) + )); + assert!(matches!( + parse_scheduled_disposition("NO_REPLY[REFUSE]: denied"), + ScheduledDisposition::Refused(_) + )); + assert!(matches!( + parse_scheduled_disposition("text mentioning NO_REPLY"), + ScheduledDisposition::Content(_) + )); + assert!(matches!( + parse_scheduled_disposition("NO_REPLY[INFO] but this is content"), + ScheduledDisposition::Content(_) + )); + assert!(matches!( + parse_scheduled_disposition(""), + ScheduledDisposition::ReportedFailure(_) + )); + } + #[test] fn test_next_run_cron_timezone_uses_from_argument() { let expr = "0 0 9 * * *".to_string(); diff --git a/src/session/messenger.rs b/src/session/messenger.rs index 104c161..d527f1d 100644 --- a/src/session/messenger.rs +++ b/src/session/messenger.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use crate::bus::{ChatMessage, MediaItem, MessageSource, OutboundMessage}; +use crate::bus::{ChatMessage, MediaItem, MessageSource, OutboundMessage, SourceKind}; use crate::session::UnifiedSessionId; use crate::tools::OutboundMessenger; @@ -77,6 +77,35 @@ impl OutboundMessenger for SessionManager { } } +impl SessionManager { + pub async fn deliver_scheduled_message( + &self, + channel: &str, + chat_id: &str, + job_id: &str, + job_name: &str, + content: &str, + ) -> Result<(), String> { + ::send_message( + self, + channel, + chat_id, + None, + content, + MessageSource { + kind: SourceKind::ExternalTrigger, + from_channel: Some("scheduler".to_string()), + from_session: Some(format!("cron:{job_id}")), + from_user_id: None, + system_name: Some(job_name.to_string()), + task_id: Some(job_id.to_string()), + }, + Vec::new(), + ) + .await + } +} + fn outbound_history_message( content: impl Into, source: MessageSource, diff --git a/src/session/session.rs b/src/session/session.rs index 0dd35a4..5c07c05 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -1119,6 +1119,29 @@ impl SessionManager { .with_context_window(self.provider_config.token_limit)) } + fn create_managed_scheduled_agent(&self) -> Result<(AgentLoop, Arc), AgentError> { + let tools = self.tools.without(&[ + "send_message", + "cron_add", + "cron_update", + "cron_remove", + "cron_enable", + "cron_disable", + ]); + let provider = create_provider(self.provider_config.clone()) + .map_err(|e| AgentError::Other(format!("failed to create scheduled provider: {e}")))?; + let agent = AgentLoop::with_provider_and_tools( + Arc::from(provider), + tools.clone(), + self.provider_config.max_tool_iterations, + self.provider_config.model_id.clone(), + self.provider_config.workspace_dir.clone(), + self.provider_config.input_types.clone(), + ) + .with_context_window(self.provider_config.token_limit); + Ok((agent, tools)) + } + /// 获取所有可用的斜杠命令 pub fn get_slash_commands(&self) -> &[SlashCommand] { SLASH_COMMANDS @@ -2506,6 +2529,38 @@ impl SessionManager { Ok(HandleResult::AgentResponse(result.final_response.content)) } + /// Execute a scheduler-managed task. The agent returns a result but cannot + /// deliver it itself; Scheduler applies the configured delivery policy. + pub async fn handle_managed_scheduled_message( + &self, + prompt: &str, + job_id: &str, + job_name: &str, + monitor: bool, + ) -> Result { + let (agent, tools) = self.create_managed_scheduled_agent()?; + let base_prompt = build_system_prompt( + &self.provider_config.workspace_dir, + &self.provider_config.model_id, + &tools, + ); + let skills_prompt = self.skills_loader.build_skills_prompt(); + let result_contract = if monitor { + "这是无人值守巡检。完成必要检查后：一切正常且无需用户关注时，只返回 NO_REPLY[INFO]: <简短原因>；发现问题时返回简洁、可操作的告警；无法完成时返回 NO_REPLY[FAIL]: <原因>；因安全或权限拒绝时返回 NO_REPLY[REFUSE]: <原因>。不要调用 send_message，不要把不确定当作正常。" + } else { + "这是 Scheduler 托管投递的定时任务。完成任务后只返回应交付给用户的最终内容，不要调用 send_message。" + }; + let system = format!( + "{base_prompt}\n\n{skills_prompt}\n\n## 定时任务执行\n任务「{job_name}」({job_id})。\n{result_contract}" + ); + let history = vec![ChatMessage::system(system), ChatMessage::user(prompt)]; + let source_session = format!("cron:{job_id}"); + let result = CURRENT_SOURCE_SESSION + .scope(Some(source_session), async { agent.process(history).await }) + .await?; + Ok(result.final_response.content) + } + pub async fn clear_session_history( &self, unified_id: &UnifiedSessionId, diff --git a/src/storage/mod.rs b/src/storage/mod.rs index edfb719..a44fdd2 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -7,14 +7,14 @@ pub mod session; pub use background_task::BackgroundTask; pub use error::StorageError; -pub use scheduler::{JobRun, ScheduledJob}; +pub use scheduler::{DeliveryPolicy, JobKind, JobRun, ScheduledJob}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; use sqlx::{Pool, Row, Sqlite}; use std::path::Path; use tokio::time::{Duration, sleep}; -const SCHEMA_VERSION: i64 = 1; +const SCHEMA_VERSION: i64 = 2; const INSERT_MESSAGE_SQL: &str = r#" INSERT INTO messages (id, session_id, seq, role, content, reasoning_content, media_refs, tool_call_id, tool_name, tool_calls, source, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) @@ -307,6 +307,19 @@ impl Storage { ("scheduled_jobs", "locked_at", "locked_at INTEGER"), ("scheduled_jobs", "lock_owner", "lock_owner TEXT"), ("scheduled_jobs", "lease_until", "lease_until INTEGER"), + ( + "scheduled_jobs", + "job_kind", + "job_kind TEXT NOT NULL DEFAULT 'task'", + ), + ( + "scheduled_jobs", + "delivery_policy", + "delivery_policy TEXT NOT NULL DEFAULT 'direct'", + ), + ("job_runs", "result_kind", "result_kind TEXT"), + ("job_runs", "delivery_status", "delivery_status TEXT"), + ("job_runs", "delivery_error", "delivery_error TEXT"), ] { let pragma = format!("PRAGMA table_info({table})"); let columns = sqlx::query(&pragma).fetch_all(&mut *tx).await?; @@ -365,6 +378,8 @@ impl Storage { channel TEXT NOT NULL, chat_id TEXT NOT NULL, model TEXT, + job_kind TEXT NOT NULL DEFAULT 'task', + delivery_policy TEXT NOT NULL DEFAULT 'direct', enabled INTEGER NOT NULL DEFAULT 1, delete_after_run INTEGER NOT NULL DEFAULT 0, next_run_at INTEGER NOT NULL, @@ -392,7 +407,10 @@ impl Storage { status TEXT NOT NULL, output TEXT, error TEXT, - duration_ms INTEGER NOT NULL + duration_ms INTEGER NOT NULL, + result_kind TEXT, + delivery_status TEXT, + delivery_error TEXT ) "#, ) diff --git a/src/storage/scheduler.rs b/src/storage/scheduler.rs index 464a5d5..a294046 100644 --- a/src/storage/scheduler.rs +++ b/src/storage/scheduler.rs @@ -4,6 +4,42 @@ use sqlx::Row; use crate::scheduler::Schedule; use crate::storage::StorageError; +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobKind { + Task, + Monitor, +} + +impl JobKind { + pub fn as_str(self) -> &'static str { + match self { + Self::Task => "task", + Self::Monitor => "monitor", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DeliveryPolicy { + Direct, + Always, + OnAlert, + Never, +} + +impl DeliveryPolicy { + pub fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Always => "always", + Self::OnAlert => "on_alert", + Self::Never => "never", + } + } +} + /// A scheduled job stored in the database. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ScheduledJob { @@ -15,6 +51,8 @@ pub struct ScheduledJob { pub channel: String, pub chat_id: String, pub model: Option, + pub job_kind: JobKind, + pub delivery_policy: DeliveryPolicy, pub enabled: bool, pub delete_after_run: bool, pub next_run_at: i64, @@ -36,9 +74,67 @@ pub struct JobRun { pub output: Option, pub error: Option, pub duration_ms: i64, + pub result_kind: Option, + pub delivery_status: Option, + pub delivery_error: Option, } impl crate::storage::Storage { + /// Seed the built-in maintenance monitor once. `INSERT OR IGNORE` preserves + /// user changes such as disabling or editing the task. + pub async fn ensure_default_maintenance_job(&self) -> Result<(), StorageError> { + let now = now_ms(); + let job = ScheduledJob { + id: "picobot-routine-maintenance".to_string(), + name: "PicoBot 日常维护巡检".to_string(), + schedule: Schedule::Every { every_ms: 86_400_000 }, + prompt: "调用 routine_maintenance 工具恰好一次。检查工具结果；成功时返回 NO_REPLY[INFO]: 日常维护完成；工具失败或结果不完整时返回 NO_REPLY[FAIL]: <原因>。不要删除 knowledge 类型的长期记忆。".to_string(), + channel: "cli_chat".to_string(), + chat_id: "maintenance".to_string(), + model: None, + job_kind: JobKind::Monitor, + delivery_policy: DeliveryPolicy::Never, + enabled: true, + delete_after_run: false, + next_run_at: now.saturating_add(30 * 60 * 1000), + last_run_at: None, + last_status: None, + last_error: None, + created_at: now, + updated_at: now, + }; + let schedule_json = serialize_schedule(&job.schedule)?; + sqlx::query( + r#" + INSERT OR IGNORE INTO scheduled_jobs + (id, name, schedule, prompt, channel, chat_id, model, job_kind, + delivery_policy, enabled, delete_after_run, next_run_at, + last_run_at, last_status, last_error, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&job.id) + .bind(&job.name) + .bind(schedule_json) + .bind(&job.prompt) + .bind(&job.channel) + .bind(&job.chat_id) + .bind(&job.model) + .bind(job.job_kind.as_str()) + .bind(job.delivery_policy.as_str()) + .bind(job.enabled as i32) + .bind(job.delete_after_run as i32) + .bind(job.next_run_at) + .bind(job.last_run_at) + .bind(&job.last_status) + .bind(&job.last_error) + .bind(job.created_at) + .bind(job.updated_at) + .execute(self.pool()) + .await?; + Ok(()) + } + /// Insert a new scheduled job. pub async fn add_scheduled_job(&self, job: &ScheduledJob) -> Result<(), StorageError> { let schedule_json = serialize_schedule(&job.schedule)?; @@ -46,9 +142,9 @@ impl crate::storage::Storage { r#" INSERT INTO scheduled_jobs (id, name, schedule, prompt, channel, chat_id, model, - enabled, delete_after_run, next_run_at, last_run_at, + job_kind, delivery_policy, enabled, delete_after_run, next_run_at, last_run_at, last_status, last_error, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&job.id) @@ -58,6 +154,8 @@ impl crate::storage::Storage { .bind(&job.channel) .bind(&job.chat_id) .bind(&job.model) + .bind(job.job_kind.as_str()) + .bind(job.delivery_policy.as_str()) .bind(job.enabled as i32) .bind(job.delete_after_run as i32) .bind(job.next_run_at) @@ -113,6 +211,25 @@ impl crate::storage::Storage { Ok(()) } + pub async fn set_scheduled_job_behavior( + &self, + id: &str, + job_kind: Option, + delivery_policy: Option, + ) -> Result<(), StorageError> { + let current = self.get_scheduled_job(id).await?; + sqlx::query( + "UPDATE scheduled_jobs SET job_kind = ?, delivery_policy = ?, updated_at = ? WHERE id = ?", + ) + .bind(job_kind.unwrap_or(current.job_kind).as_str()) + .bind(delivery_policy.unwrap_or(current.delivery_policy).as_str()) + .bind(now_ms()) + .bind(id) + .execute(self.pool()) + .await?; + Ok(()) + } + /// Update selective fields on a scheduled job. pub async fn update_scheduled_job( &self, @@ -258,8 +375,9 @@ impl crate::storage::Storage { } else { sqlx::query( r#" - INSERT INTO job_runs (job_id, started_at, finished_at, status, output, error, duration_ms) - VALUES (?, ?, ?, ?, ?, ?, ?) + INSERT INTO job_runs (job_id, started_at, finished_at, status, output, error, duration_ms, + result_kind, delivery_status, delivery_error) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&run.job_id) @@ -269,6 +387,9 @@ impl crate::storage::Storage { .bind(&run.output) .bind(&run.error) .bind(run.duration_ms) + .bind(&run.result_kind) + .bind(&run.delivery_status) + .bind(&run.delivery_error) .execute(&mut *tx) .await?; @@ -343,6 +464,9 @@ impl crate::storage::Storage { output: r.try_get("output")?, error: r.try_get("error")?, duration_ms: r.try_get("duration_ms")?, + result_kind: r.try_get("result_kind")?, + delivery_status: r.try_get("delivery_status")?, + delivery_error: r.try_get("delivery_error")?, }) }) .collect() @@ -381,6 +505,8 @@ fn row_to_job(row: &sqlx::sqlite::SqliteRow) -> Result("job_kind")?)?, + delivery_policy: parse_delivery_policy(&row.try_get::("delivery_policy")?)?, enabled: row.try_get::("enabled")? != 0, delete_after_run: row.try_get::("delete_after_run")? != 0, next_run_at: row.try_get("next_run_at")?, @@ -392,9 +518,31 @@ fn row_to_job(row: &sqlx::sqlite::SqliteRow) -> Result Result { + match value { + "task" => Ok(JobKind::Task), + "monitor" => Ok(JobKind::Monitor), + other => Err(StorageError::Serialization(format!( + "unknown job kind: {other}" + ))), + } +} + +fn parse_delivery_policy(value: &str) -> Result { + match value { + "direct" => Ok(DeliveryPolicy::Direct), + "always" => Ok(DeliveryPolicy::Always), + "on_alert" => Ok(DeliveryPolicy::OnAlert), + "never" => Ok(DeliveryPolicy::Never), + other => Err(StorageError::Serialization(format!( + "unknown delivery policy: {other}" + ))), + } +} + #[cfg(test)] mod tests { - use super::ScheduledJob; + use super::{DeliveryPolicy, JobKind, ScheduledJob}; use crate::scheduler::Schedule; use crate::storage::Storage; use sqlx::SqlitePool; @@ -425,6 +573,32 @@ mod tests { assert_eq!(row.0, 0); } + #[tokio::test] + async fn default_maintenance_job_is_idempotent_and_preserves_user_state() { + let storage = setup_storage().await; + storage.ensure_default_maintenance_job().await.unwrap(); + let job = storage + .get_scheduled_job("picobot-routine-maintenance") + .await + .unwrap(); + assert_eq!(job.job_kind, JobKind::Monitor); + assert_eq!(job.delivery_policy, DeliveryPolicy::Never); + assert!(job.enabled); + + storage + .set_scheduled_job_enabled("picobot-routine-maintenance", false) + .await + .unwrap(); + storage.ensure_default_maintenance_job().await.unwrap(); + assert!( + !storage + .get_scheduled_job("picobot-routine-maintenance") + .await + .unwrap() + .enabled + ); + } + #[tokio::test] async fn test_add_and_get_job() { let storage = setup_storage().await; @@ -437,6 +611,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "conn-1".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t + 3600000, @@ -466,6 +642,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "conn-1".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t + 1000, @@ -493,6 +671,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -520,6 +700,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -550,6 +732,8 @@ mod tests { channel: "feishu".into(), chat_id: "oc_1".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -587,6 +771,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -628,6 +814,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -651,6 +839,9 @@ mod tests { output: Some("done".into()), error: None, duration_ms: 10, + result_kind: Some("content".into()), + delivery_status: Some("direct".into()), + delivery_error: None, }; storage .complete_scheduled_job(&run, "owner", Some(t + 2000), false, false) diff --git a/src/tools/cron.rs b/src/tools/cron.rs index d96cb43..78ea5c9 100644 --- a/src/tools/cron.rs +++ b/src/tools/cron.rs @@ -5,7 +5,7 @@ use serde_json::{Value, json}; use uuid::Uuid; use crate::scheduler::{Schedule, next_run_for_schedule}; -use crate::storage::{ScheduledJob, Storage}; +use crate::storage::{DeliveryPolicy, JobKind, ScheduledJob, Storage}; use crate::tools::traits::{Tool, ToolResult}; fn now_ms() -> i64 { @@ -75,6 +75,16 @@ impl Tool for CronAddTool { "model": { "type": "string", "description": "Optional model override for this job" + }, + "kind": { + "type": "string", + "enum": ["task", "monitor"], + "description": "task always reports a result; monitor is quiet when healthy" + }, + "delivery_policy": { + "type": "string", + "enum": ["always", "on_alert", "never"], + "description": "Scheduler-managed delivery policy. Defaults to always for task and on_alert for monitor." } }, "required": ["schedule", "prompt", "channel", "chat_id"] @@ -152,6 +162,38 @@ impl Tool for CronAddTool { .get("model") .and_then(|v| v.as_str()) .map(|s| s.to_string()); + let job_kind = match args.get("kind").and_then(|v| v.as_str()).unwrap_or("task") { + "task" => JobKind::Task, + "monitor" => JobKind::Monitor, + value => { + return Ok(ToolResult { + success: false, + output: String::new(), + error: Some(format!("invalid kind: {value}")), + }); + } + }; + let default_policy = if job_kind == JobKind::Monitor { + "on_alert" + } else { + "always" + }; + let delivery_policy = match args + .get("delivery_policy") + .and_then(|v| v.as_str()) + .unwrap_or(default_policy) + { + "always" => DeliveryPolicy::Always, + "on_alert" => DeliveryPolicy::OnAlert, + "never" => DeliveryPolicy::Never, + value => { + return Ok(ToolResult { + success: false, + output: String::new(), + error: Some(format!("invalid delivery_policy: {value}")), + }); + } + }; let now = now_ms(); let next_run_at = next_run_for_schedule(&schedule, now) @@ -166,6 +208,8 @@ impl Tool for CronAddTool { channel, chat_id, model, + job_kind, + delivery_policy, enabled: true, delete_after_run: false, next_run_at, @@ -256,8 +300,17 @@ impl Tool for CronListTool { }; let model = j.model.as_deref().unwrap_or("default"); lines.push(format!( - "[{}] id={} name=\"{}\" channel={} chat={} model={} next={}{}", - status, j.id, j.name, j.channel, j.chat_id, model, j.next_run_at, last + "[{}] id={} name=\"{}\" kind={} delivery={} channel={} chat={} model={} next={}{}", + status, + j.id, + j.name, + j.job_kind.as_str(), + j.delivery_policy.as_str(), + j.channel, + j.chat_id, + model, + j.next_run_at, + last )); } @@ -524,6 +577,14 @@ impl Tool for CronUpdateTool { "model": { "type": "string", "description": "New model override" + }, + "kind": { + "type": "string", + "enum": ["task", "monitor"] + }, + "delivery_policy": { + "type": "string", + "enum": ["always", "on_alert", "never"] } }, "required": ["job_id"] @@ -573,10 +634,38 @@ impl Tool for CronUpdateTool { .get("model") .and_then(|v| v.as_str()) .map(|s| s.to_string()); + let job_kind = match args.get("kind").and_then(|v| v.as_str()) { + Some("task") => Some(JobKind::Task), + Some("monitor") => Some(JobKind::Monitor), + Some(value) => { + return Ok(ToolResult { + success: false, + output: String::new(), + error: Some(format!("invalid kind: {value}")), + }); + } + None => None, + }; + let delivery_policy = match args.get("delivery_policy").and_then(|v| v.as_str()) { + Some("always") => Some(DeliveryPolicy::Always), + Some("on_alert") => Some(DeliveryPolicy::OnAlert), + Some("never") => Some(DeliveryPolicy::Never), + Some(value) => { + return Ok(ToolResult { + success: false, + output: String::new(), + error: Some(format!("invalid delivery_policy: {value}")), + }); + } + None => None, + }; self.storage .update_scheduled_job(&job_id, prompt, schedule, channel, chat_id, model) .await?; + self.storage + .set_scheduled_job_behavior(&job_id, job_kind, delivery_policy) + .await?; if args.get("schedule").is_some() { let job = self.storage.get_scheduled_job(&job_id).await?; @@ -636,6 +725,28 @@ mod tests { let jobs = storage.list_scheduled_jobs().await.unwrap(); assert_eq!(jobs.len(), 1); assert_eq!(jobs[0].name, "hourly report"); + assert_eq!(jobs[0].job_kind, JobKind::Task); + assert_eq!(jobs[0].delivery_policy, DeliveryPolicy::Always); + } + + #[tokio::test] + async fn monitor_defaults_to_alert_only_delivery() { + let storage = setup_storage().await; + let tool = CronAddTool::new(storage.clone(), vec!["cli_chat".to_string()]); + let result = tool + .execute(json!({ + "schedule": {"type": "every", "every_ms": 3600000}, + "prompt": "check health", + "channel": "cli_chat", + "chat_id": "test-chat-1", + "kind": "monitor" + })) + .await + .unwrap(); + assert!(result.success); + let jobs = storage.list_scheduled_jobs().await.unwrap(); + assert_eq!(jobs[0].job_kind, JobKind::Monitor); + assert_eq!(jobs[0].delivery_policy, DeliveryPolicy::OnAlert); } #[tokio::test] @@ -668,6 +779,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t + 1000, @@ -697,6 +810,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -729,6 +844,8 @@ mod tests { channel: "cli_chat".into(), chat_id: "c".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t, @@ -773,6 +890,8 @@ mod tests { channel: "feishu".into(), chat_id: "oc_1".into(), model: None, + job_kind: JobKind::Task, + delivery_policy: DeliveryPolicy::Direct, enabled: true, delete_after_run: false, next_run_at: t + 1000, diff --git a/src/tools/maintenance.rs b/src/tools/maintenance.rs new file mode 100644 index 0000000..b4a3c37 --- /dev/null +++ b/src/tools/maintenance.rs @@ -0,0 +1,52 @@ +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::storage::Storage; + +use super::traits::{Tool, ToolResult}; + +/// Safe, bounded maintenance operations used by the built-in monitor job. +pub struct RoutineMaintenanceTool { + storage: Arc, + timeline_retention_days: u64, +} + +impl RoutineMaintenanceTool { + pub fn new(storage: Arc, timeline_retention_days: u64) -> Self { + Self { + storage, + timeline_retention_days, + } + } +} + +#[async_trait] +impl Tool for RoutineMaintenanceTool { + fn name(&self) -> &str { + "routine_maintenance" + } + + fn description(&self) -> &str { + "Run PicoBot's safe routine maintenance: remove timeline memories older than the configured retention period. It never deletes long-term knowledge memories." + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({"type": "object", "properties": {}}) + } + + async fn execute(&self, _args: serde_json::Value) -> anyhow::Result { + let removed = self + .storage + .cleanup_old_timelines(self.timeline_retention_days) + .await?; + Ok(ToolResult { + success: true, + output: format!( + "Routine maintenance completed: removed {removed} timeline memories older than {} days; knowledge memories were not modified.", + self.timeline_retention_days + ), + error: None, + }) + } +} diff --git a/src/tools/mod.rs b/src/tools/mod.rs index 2266f9c..4b6379b 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -12,6 +12,7 @@ pub mod file_search; pub mod file_write; pub mod get_skill; pub mod http_request; +pub mod maintenance; pub mod memory; pub mod path_utils; pub mod pty; @@ -33,6 +34,7 @@ pub use file_search::FileSearchTool; pub use file_write::FileWriteTool; pub use get_skill::GetSkillTool; pub use http_request::HttpRequestTool; +pub use maintenance::RoutineMaintenanceTool; pub use memory::{MemoryForgetTool, MemoryRecallTool, MemoryStoreTool, TimelineRecallTool}; pub use pty::{PtyManager, PtyTool}; pub use registry::ToolRegistry; diff --git a/src/tools/registry.rs b/src/tools/registry.rs index 57b96c1..9a07390 100644 --- a/src/tools/registry.rs +++ b/src/tools/registry.rs @@ -75,6 +75,18 @@ impl ToolRegistry { .collect() } + /// Clone this registry while excluding tools that a constrained execution + /// path must not call (for example scheduler-managed delivery). + pub fn without(&self, excluded: &[&str]) -> Arc { + let filtered = Self::new(); + for (name, tool) in self.iter() { + if !excluded.contains(&name.as_str()) { + filtered.register_raw(name, tool); + } + } + Arc::new(filtered) + } + /// 生成工具列表描述，用于子 Agent 系统提示词 pub fn describe_for_prompt(&self) -> String { let mut entries: Vec = self diff --git a/webui/src/pages/TasksPage.svelte b/webui/src/pages/TasksPage.svelte index fe2a308..55d58d3 100644 --- a/webui/src/pages/TasksPage.svelte +++ b/webui/src/pages/TasksPage.svelte @@ -59,8 +59,8 @@ {:else} {#each jobs as job (job.id)}
-

{job.name}

{job.prompt}

{job.channel} · {job.chat_id}下次 {formatTime(job.next_run_at)}上次 {formatTime(job.last_run_at)}
- {#if runs[job.id]?.length}
{#each runs[job.id] as run}
{formatTime(run.finished_at)} · {run.duration_ms}ms
{/each}
{/if} +

{job.name}

{job.prompt}

{job.job_kind === "monitor" ? "巡检" : "任务"} · {job.delivery_policy}{job.channel} · {job.chat_id}下次 {formatTime(job.next_run_at)}上次 {formatTime(job.last_run_at)}
+ {#if runs[job.id]?.length}
{#each runs[job.id] as run}
{formatTime(run.finished_at)} · {run.duration_ms}ms{run.result_kind ? ` · ${run.result_kind}` : ""}{run.delivery_status ? ` · ${run.delivery_status}` : ""}
{/each}
{/if}
{:else}
暂无定时任务
{/each} {/if}