扩充定时任务功能、实现类似其他agent软件的heartbeat功能

This commit is contained in:
xiaoxixi 2026-07-15 16:36:10 +08:00
parent 6e56a84054
commit 9ac898acbb
16 changed files with 753 additions and 58 deletions

View File

@ -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

View File

@ -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` 清理过期 TimelineKnowledge 不会被自动删除。
### 工具
@ -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 动态发现并注册 |

View File

@ -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 和 SessionManagerScheduler 启用时幂等创建默认日常维护巡检
4. 注册内置工具、渠道、MCP 工具和 Cron 工具。
5. 启动所有 Channel。
6. 通过 TaskSupervisor 启动 message processor、dispatcher 和 scheduler。

View File

@ -70,7 +70,7 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取
| `enabled` | bool | true | 是否启动调度器并注册 cron 工具 |
| `poll_interval_secs` | int | 60 | 检查到期任务的轮询间隔 |
| `max_concurrent` | int | 1 | 每批到期任务的最大并发数,运行时限制在 1256 |
| `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 | 默认日常维护巡检删除超过该期限的 TimelineKnowledge 不受影响 |
| `max_failures_before_degrade` | int | 3 | 预留的归并失败阈值;当前无失败降级循环 |
注意:这些字段都会被解析,但当前 worker 的 Knowledge 召回数量仍固定为 5idle consolidation、Timeline 自动清理和失败降级循环尚未接入。配置存在不等于对应后台行为已经生效
注意:当前 worker 的 Knowledge 召回数量仍固定为 5idle consolidation 和失败降级循环尚未接入。Timeline 清理由默认启用的 `picobot-routine-maintenance` 定时巡检执行
## channels.feishu 字段

View File

@ -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

View File

@ -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()

View File

@ -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<i64> {
@ -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, &notice)
.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();

View File

@ -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> {
<Self as OutboundMessenger>::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<String>,
source: MessageSource,

View File

@ -1119,6 +1119,29 @@ impl SessionManager {
.with_context_window(self.provider_config.token_limit))
}
fn create_managed_scheduled_agent(&self) -> Result<(AgentLoop, Arc<ToolRegistry>), 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<String, AgentError> {
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,

View File

@ -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
)
"#,
)

View File

@ -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<String>,
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<String>,
pub error: Option<String>,
pub duration_ms: i64,
pub result_kind: Option<String>,
pub delivery_status: Option<String>,
pub delivery_error: Option<String>,
}
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<JobKind>,
delivery_policy: Option<DeliveryPolicy>,
) -> 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<ScheduledJob, StorageErro
channel: row.try_get("channel")?,
chat_id: row.try_get("chat_id")?,
model: row.try_get("model")?,
job_kind: parse_job_kind(&row.try_get::<String, _>("job_kind")?)?,
delivery_policy: parse_delivery_policy(&row.try_get::<String, _>("delivery_policy")?)?,
enabled: row.try_get::<i32, _>("enabled")? != 0,
delete_after_run: row.try_get::<i32, _>("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<ScheduledJob, StorageErro
})
}
fn parse_job_kind(value: &str) -> Result<JobKind, StorageError> {
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<DeliveryPolicy, StorageError> {
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)

View File

@ -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,

52
src/tools/maintenance.rs Normal file
View File

@ -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<Storage>,
timeline_retention_days: u64,
}
impl RoutineMaintenanceTool {
pub fn new(storage: Arc<Storage>, 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<ToolResult> {
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,
})
}
}

View File

@ -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;

View File

@ -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<Self> {
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<String> = self

View File

@ -59,8 +59,8 @@
{:else}
{#each jobs as job (job.id)}
<article class="card">
<div class="card-row"><div><h3>{job.name}</h3><p>{job.prompt}</p><div class="meta"><span>{job.channel} · {job.chat_id}</span><span>下次 {formatTime(job.next_run_at)}</span><span>上次 {formatTime(job.last_run_at)}</span></div></div><StatusBadge status={job.enabled ? (job.last_status || "enabled") : "disabled"} /></div>
{#if runs[job.id]?.length}<div class="details">{#each runs[job.id] as run}<div class="card-row run-row"><span class="meta">{formatTime(run.finished_at)} · {run.duration_ms}ms</span><StatusBadge status={run.status} /></div>{/each}</div>{/if}
<div class="card-row"><div><h3>{job.name}</h3><p>{job.prompt}</p><div class="meta"><span>{job.job_kind === "monitor" ? "巡检" : "任务"} · {job.delivery_policy}</span><span>{job.channel} · {job.chat_id}</span><span>下次 {formatTime(job.next_run_at)}</span><span>上次 {formatTime(job.last_run_at)}</span></div></div><StatusBadge status={job.enabled ? (job.last_status || "enabled") : "disabled"} /></div>
{#if runs[job.id]?.length}<div class="details">{#each runs[job.id] as run}<div class="card-row run-row"><span class="meta">{formatTime(run.finished_at)} · {run.duration_ms}ms{run.result_kind ? ` · ${run.result_kind}` : ""}{run.delivery_status ? ` · ${run.delivery_status}` : ""}</span><StatusBadge status={run.status} /></div>{/each}</div>{/if}
</article>
{:else}<div class="empty-card">暂无定时任务</div>{/each}
{/if}