nexo_core/agent/
cron_tool.rs

1//! `cron_create` / `cron_list` / `cron_delete` LLM tools that
2//! read/write the [`CronStore`].
3//!
4//! Reference (PRIMARY):
5//!   * `claude-code-leak/src/tools/ScheduleCronTool/CronCreateTool.ts:1-157`
6//!     (5-field cron schema, recurring + durable flags, 50-entry
7//!     cap).
8//!   * `claude-code-leak/src/tools/ScheduleCronTool/CronListTool.ts`
9//!     and `CronDeleteTool.ts` (sibling tools — same store).
10//!
11//! Reference (secondary):
12//!   * OpenClaw `research/src/cron/schedule.ts` —
13//!     `croner` + cache. Cron expression semantics are
14//!     compatible.
15//!
16//! Runtime firing is shipped: `CronRunner` polls `due_at` and
17//! dispatches due entries through `LlmCronDispatcher`.
18
19use super::context::AgentContext;
20use super::tool_registry::ToolHandler;
21use crate::cron_schedule::{
22    build_new_entry, next_fire_after, CronStore, CronStoreError, MAX_CRON_ENTRIES_PER_BINDING,
23};
24use async_trait::async_trait;
25use nexo_llm::ToolDef;
26use serde_json::{json, Value};
27use std::sync::Arc;
28
29/// `cron_create` — schedule a recurring (or one-shot) prompt.
30pub struct CronCreateTool {
31    store: Arc<dyn CronStore>,
32}
33
34impl CronCreateTool {
35    pub fn new(store: Arc<dyn CronStore>) -> Self {
36        Self { store }
37    }
38
39    pub fn tool_def() -> ToolDef {
40        ToolDef {
41            name: "cron_create".to_string(),
42            description: format!(
43                "Schedule a recurring or one-shot prompt to fire on a cron schedule. The runtime persists the entry to SQLite and survives daemon restarts. Entries are namespaced to the originating binding (`plugin:instance` from inbound origin; fallback `agent_id`). Cap: {} entries per binding. Minimum interval: 60 seconds. One-shot entries auto-delete on success; on dispatch failure they retry with bounded backoff per `runtime.cron.one_shot_retry` before final drop.",
44                MAX_CRON_ENTRIES_PER_BINDING
45            ),
46            parameters: json!({
47                "type": "object",
48                "properties": {
49                    "cron": {
50                        "type": "string",
51                        "description": "Standard cron expression in UTC. Prefer 5-field: \"M H DoM Mon DoW\" (6-field with seconds is also accepted). Examples: \"*/5 * * * *\" (every 5 minutes), \"0 9 * * *\" (daily 9am UTC), \"30 14 28 2 *\" (Feb 28 14:30 UTC, runs once if recurring=false). Expressions that fire more often than once per 60 seconds are rejected."
52                    },
53                    "prompt": {
54                        "type": "string",
55                        "description": "Prompt to enqueue at each fire time."
56                    },
57                    "channel": {
58                        "type": "string",
59                        "description": "Optional channel hint (e.g. 'whatsapp:default'). Used only when paired with `recipient` for outbound delivery; otherwise the cron fire is log-only."
60                    },
61                    "recipient": {
62                        "type": "string",
63                        "description": "Optional channel-specific recipient id (WhatsApp JID, Telegram chat_id, email address). When set together with `channel`, the runtime routes the model's response back to this recipient on every fire. Without it, fires log only."
64                    },
65                    "recurring": {
66                        "type": "boolean",
67                        "description": "true (default) = fire on every cron match until deleted. false = fire once at the next match, then auto-delete (use for 'remind me at X')."
68                    }
69                },
70                "required": ["cron", "prompt"]
71            }),
72        }
73    }
74}
75
76fn binding_id_from_ctx(ctx: &AgentContext) -> String {
77    ctx.inbound_origin
78        .as_ref()
79        .map(|(plugin, instance, _sender)| format!("{plugin}:{instance}"))
80        .unwrap_or_else(|| ctx.agent_id.clone())
81}
82
83#[async_trait]
84impl ToolHandler for CronCreateTool {
85    async fn call(&self, ctx: &AgentContext, args: Value) -> anyhow::Result<Value> {
86        let cron = args
87            .get("cron")
88            .and_then(|v| v.as_str())
89            .ok_or_else(|| anyhow::anyhow!("cron_create requires `cron` (string)"))?
90            .trim()
91            .to_string();
92        let prompt = args
93            .get("prompt")
94            .and_then(|v| v.as_str())
95            .ok_or_else(|| anyhow::anyhow!("cron_create requires `prompt` (string)"))?
96            .to_string();
97        let channel = args
98            .get("channel")
99            .and_then(|v| v.as_str())
100            .map(str::to_string);
101        let recipient = args
102            .get("recipient")
103            .and_then(|v| v.as_str())
104            .map(str::to_string);
105        let recurring = args
106            .get("recurring")
107            .and_then(|v| v.as_bool())
108            .unwrap_or(true);
109
110        let binding_id = binding_id_from_ctx(ctx);
111        let effective = ctx.effective_policy();
112        let model_provider = effective.model.provider.trim();
113        let model_name = effective.model.model.trim();
114        let entry = build_new_entry(
115            &self.store,
116            &binding_id,
117            &cron,
118            &prompt,
119            channel.as_deref(),
120            recurring,
121            recipient.as_deref(),
122            if model_provider.is_empty() {
123                None
124            } else {
125                Some(model_provider)
126            },
127            if model_name.is_empty() {
128                None
129            } else {
130                Some(model_name)
131            },
132            // Pin the agent's tenant on every cron entry so the
133            // cron LLM dispatcher resolves the tenant's own
134            // provider keys.
135            ctx.config.tenant_id.as_deref(),
136        )
137        .await
138        .map_err(map_err)?;
139        let id = entry.id.clone();
140        let next_fire_at = entry.next_fire_at;
141        self.store.insert(&entry).await.map_err(map_err)?;
142        Ok(json!({
143            "ok": true,
144            "id": id,
145            "binding_id": binding_id,
146            "cron": cron,
147            "recurring": recurring,
148            "next_fire_at": next_fire_at,
149            "model_provider": entry.model_provider,
150            "model_name": entry.model_name,
151            "instructions": "Entry persisted. The runtime fires it on schedule. Use cron_list to inspect, cron_pause/cron_resume to temporarily stop/restart, and cron_delete to cancel."
152        }))
153    }
154}
155
156/// `cron_list` — return the binding's scheduled entries.
157pub struct CronListTool {
158    store: Arc<dyn CronStore>,
159}
160
161impl CronListTool {
162    pub fn new(store: Arc<dyn CronStore>) -> Self {
163        Self { store }
164    }
165
166    pub fn tool_def() -> ToolDef {
167        ToolDef {
168            name: "cron_list".to_string(),
169            description: "List scheduled cron entries for the current binding namespace (origin-tagged `plugin:instance`, or `agent_id` fallback). Read-only."
170                .to_string(),
171            parameters: json!({
172                "type": "object",
173                "properties": {},
174                "required": []
175            }),
176        }
177    }
178}
179
180#[async_trait]
181impl ToolHandler for CronListTool {
182    async fn call(&self, ctx: &AgentContext, _args: Value) -> anyhow::Result<Value> {
183        let binding_id = binding_id_from_ctx(ctx);
184        let entries = self
185            .store
186            .list_by_binding(&binding_id)
187            .await
188            .map_err(map_err)?;
189        Ok(json!({
190            "binding_id": binding_id,
191            "count": entries.len(),
192            "entries": entries,
193        }))
194    }
195}
196
197/// `cron_pause` / `cron_resume` — toggle the `paused` flag on an
198/// entry without dropping it. Useful to silence a recurring job
199/// during a maintenance window.
200pub struct CronPauseTool {
201    store: Arc<dyn CronStore>,
202}
203
204impl CronPauseTool {
205    pub fn new(store: Arc<dyn CronStore>) -> Self {
206        Self { store }
207    }
208    pub fn tool_def() -> ToolDef {
209        ToolDef {
210            name: "cron_pause".to_string(),
211            description: "Pause a scheduled cron entry by id. The row stays in storage (`paused=true`) and `CronRunner` skips it until `cron_resume`."
212                .to_string(),
213            parameters: json!({
214                "type": "object",
215                "properties": {
216                    "id": { "type": "string", "description": "Entry id from cron_create response or cron_list output." }
217                },
218                "required": ["id"]
219            }),
220        }
221    }
222}
223
224#[async_trait]
225impl ToolHandler for CronPauseTool {
226    async fn call(&self, _ctx: &AgentContext, args: Value) -> anyhow::Result<Value> {
227        let id = args
228            .get("id")
229            .and_then(|v| v.as_str())
230            .ok_or_else(|| anyhow::anyhow!("cron_pause requires `id` (string)"))?
231            .to_string();
232        self.store.set_paused(&id, true).await.map_err(map_err)?;
233        Ok(json!({"ok": true, "id": id, "paused": true}))
234    }
235}
236
237pub struct CronResumeTool {
238    store: Arc<dyn CronStore>,
239}
240
241impl CronResumeTool {
242    pub fn new(store: Arc<dyn CronStore>) -> Self {
243        Self { store }
244    }
245    pub fn tool_def() -> ToolDef {
246        ToolDef {
247            name: "cron_resume".to_string(),
248            description:
249                "Resume a paused cron entry by id (`paused=false`, inverse of cron_pause)."
250                    .to_string(),
251            parameters: json!({
252                "type": "object",
253                "properties": {
254                    "id": { "type": "string", "description": "Entry id from cron_create response or cron_list output." }
255                },
256                "required": ["id"]
257            }),
258        }
259    }
260}
261
262#[async_trait]
263impl ToolHandler for CronResumeTool {
264    async fn call(&self, _ctx: &AgentContext, args: Value) -> anyhow::Result<Value> {
265        let id = args
266            .get("id")
267            .and_then(|v| v.as_str())
268            .ok_or_else(|| anyhow::anyhow!("cron_resume requires `id` (string)"))?
269            .to_string();
270        self.store.set_paused(&id, false).await.map_err(map_err)?;
271        Ok(json!({"ok": true, "id": id, "paused": false}))
272    }
273}
274
275/// `cron_delete` — drop a scheduled entry by id.
276pub struct CronDeleteTool {
277    store: Arc<dyn CronStore>,
278}
279
280impl CronDeleteTool {
281    pub fn new(store: Arc<dyn CronStore>) -> Self {
282        Self { store }
283    }
284
285    pub fn tool_def() -> ToolDef {
286        ToolDef {
287            name: "cron_delete".to_string(),
288            description: "Delete a scheduled cron entry by id (works for recurring and one-shot entries). Use cron_list first to find the id."
289                .to_string(),
290            parameters: json!({
291                "type": "object",
292                "properties": {
293                    "id": {
294                        "type": "string",
295                        "description": "Entry id from cron_create response or cron_list output."
296                    }
297                },
298                "required": ["id"]
299            }),
300        }
301    }
302}
303
304#[async_trait]
305impl ToolHandler for CronDeleteTool {
306    async fn call(&self, _ctx: &AgentContext, args: Value) -> anyhow::Result<Value> {
307        let id = args
308            .get("id")
309            .and_then(|v| v.as_str())
310            .ok_or_else(|| anyhow::anyhow!("cron_delete requires `id` (string)"))?
311            .to_string();
312        self.store.delete(&id).await.map_err(map_err)?;
313        Ok(json!({"ok": true, "id": id}))
314    }
315}
316
317fn map_err(e: CronStoreError) -> anyhow::Error {
318    match e {
319        CronStoreError::InvalidCron(expr, reason) => {
320            anyhow::anyhow!("invalid cron expression `{expr}`: {reason}")
321        }
322        CronStoreError::IntervalTooShort(expr, _) => {
323            anyhow::anyhow!(
324                "cron expression `{expr}` schedules fires more often than the 60-second minimum"
325            )
326        }
327        CronStoreError::BindingFull(binding, count, max) => {
328            anyhow::anyhow!(
329                "binding `{binding}` already has {count} cron entries (max {max}) — delete one first via cron_delete"
330            )
331        }
332        CronStoreError::NotFound(id) => {
333            anyhow::anyhow!("cron entry `{id}` not found")
334        }
335        CronStoreError::Sql(s) => anyhow::anyhow!("cron store sqlx error: {s}"),
336    }
337}
338
339/// Helper exposed for tests and runtime integrations: compute the
340/// next-fire timestamp for a given expression. Re-exports the
341/// `cron_schedule` helper through the agent module so callers
342/// (poller hooks, doctor commands) have one entry point.
343pub fn next_fire_for(cron_expr: &str, from_unix: i64) -> Result<i64, CronStoreError> {
344    next_fire_after(cron_expr, from_unix)
345}
346
347#[cfg(test)]
348mod tests {
349    use super::*;
350    use crate::cron_schedule::SqliteCronStore;
351    use crate::session::SessionManager;
352    use nexo_broker::AnyBroker;
353    use nexo_config::types::agents::{
354        AgentConfig, AgentRuntimeConfig, DreamingYamlConfig, HeartbeatConfig, ModelConfig,
355        OutboundAllowlistConfig, WorkspaceGitConfig,
356    };
357
358    async fn ctx_with_origin() -> (AgentContext, Arc<dyn CronStore>) {
359        let cfg = AgentConfig {
360            id: "a".into(),
361            model: ModelConfig {
362                provider: "x".into(),
363                model: "y".into(),
364            },
365            plugins: Vec::new(),
366            heartbeat: HeartbeatConfig::default(),
367            config: AgentRuntimeConfig::default(),
368            system_prompt: String::new(),
369            workspace: String::new(),
370            skills: Vec::new(),
371            skills_dir: "./skills".into(),
372            skill_overrides: Default::default(),
373            transcripts_dir: String::new(),
374            dreaming: DreamingYamlConfig::default(),
375            workspace_git: WorkspaceGitConfig::default(),
376            tool_rate_limits: None,
377            tool_args_validation: None,
378            extra_docs: Vec::new(),
379            inbound_bindings: Vec::new(),
380            allowed_tools: Vec::new(),
381            sender_rate_limit: None,
382            allowed_delegates: Vec::new(),
383            accept_delegates_from: Vec::new(),
384            description: String::new(),
385            google_auth: None,
386            credentials: Default::default(),
387            link_understanding: serde_json::Value::Null,
388            web_search: serde_json::Value::Null,
389            pairing_policy: serde_json::Value::Null,
390            language: None,
391            locale_prompts: Default::default(),
392            outbound_allowlist: OutboundAllowlistConfig::default(),
393            context_optimization: None,
394            dispatch_policy: Default::default(),
395            plan_mode: Default::default(),
396            remote_triggers: Vec::new(),
397            lsp: nexo_config::types::lsp::LspPolicy::default(),
398            config_tool: nexo_config::types::config_tool::ConfigToolPolicy::default(),
399            team: nexo_config::types::team::TeamPolicy::default(),
400            proactive: Default::default(),
401            repl: Default::default(),
402            auto_dream: None,
403            assistant_mode: None,
404            away_summary: None,
405            brief: None,
406            channels: None,
407            auto_approve: false,
408            extract_memories: None,
409            event_subscribers: Vec::new(),
410            tenant_id: None,
411            extensions_config: std::collections::BTreeMap::new(),
412            active: true,
413        };
414        let ctx = AgentContext::new(
415            "a",
416            Arc::new(cfg),
417            AnyBroker::local(),
418            Arc::new(SessionManager::new(std::time::Duration::from_secs(60), 8)),
419        )
420        .with_inbound_origin("whatsapp", "default", "+1234");
421        let store: Arc<dyn CronStore> = Arc::new(SqliteCronStore::open_memory().await.unwrap());
422        (ctx, store)
423    }
424
425    #[tokio::test]
426    async fn create_persists_entry_with_binding_namespace() {
427        let (ctx, store) = ctx_with_origin().await;
428        let tool = CronCreateTool::new(store.clone());
429        let res = tool
430            .call(
431                &ctx,
432                json!({
433                    "cron": "*/5 * * * *",
434                    "prompt": "ping ops"
435                }),
436            )
437            .await
438            .unwrap();
439        assert_eq!(res["ok"], true);
440        assert_eq!(res["binding_id"], "whatsapp:default");
441        assert!(res["next_fire_at"].as_i64().unwrap() > 0);
442        assert_eq!(store.count_by_binding("whatsapp:default").await.unwrap(), 1);
443    }
444
445    #[tokio::test]
446    async fn create_rejects_invalid_cron() {
447        let (ctx, store) = ctx_with_origin().await;
448        let tool = CronCreateTool::new(store);
449        let err = tool
450            .call(&ctx, json!({"cron": "not a cron", "prompt": "x"}))
451            .await
452            .unwrap_err()
453            .to_string();
454        assert!(err.contains("invalid cron"), "got: {err}");
455    }
456
457    #[tokio::test]
458    async fn create_rejects_sub_minute() {
459        let (ctx, store) = ctx_with_origin().await;
460        let tool = CronCreateTool::new(store);
461        let err = tool
462            .call(&ctx, json!({"cron": "* * * * * *", "prompt": "x"}))
463            .await
464            .unwrap_err()
465            .to_string();
466        assert!(err.contains("60-second"), "got: {err}");
467    }
468
469    #[tokio::test]
470    async fn list_returns_only_current_binding_entries() {
471        let (ctx, store) = ctx_with_origin().await;
472        let create = CronCreateTool::new(store.clone());
473        // Insert in current binding.
474        create
475            .call(&ctx, json!({"cron": "*/5 * * * *", "prompt": "a"}))
476            .await
477            .unwrap();
478        create
479            .call(&ctx, json!({"cron": "0 9 * * *", "prompt": "b"}))
480            .await
481            .unwrap();
482        // Insert one in a different binding bypassing the tool —
483        // simulates another goal's data.
484        let other = build_new_entry(
485            &store,
486            "telegram:bot",
487            "0 */2 * * *",
488            "c",
489            None,
490            true,
491            None,
492            None,
493            None,
494            None,
495        )
496        .await
497        .unwrap();
498        store.insert(&other).await.unwrap();
499
500        let list = CronListTool::new(store);
501        let res = list.call(&ctx, json!({})).await.unwrap();
502        assert_eq!(res["binding_id"], "whatsapp:default");
503        assert_eq!(res["count"], 2);
504    }
505
506    #[tokio::test]
507    async fn delete_removes_entry() {
508        let (ctx, store) = ctx_with_origin().await;
509        let create = CronCreateTool::new(store.clone());
510        let res = create
511            .call(&ctx, json!({"cron": "*/5 * * * *", "prompt": "x"}))
512            .await
513            .unwrap();
514        let id = res["id"].as_str().unwrap().to_string();
515        let del = CronDeleteTool::new(store.clone());
516        let res2 = del.call(&ctx, json!({"id": id.clone()})).await.unwrap();
517        assert_eq!(res2["ok"], true);
518        assert_eq!(store.count_by_binding("whatsapp:default").await.unwrap(), 0);
519    }
520
521    #[tokio::test]
522    async fn delete_unknown_id_errors() {
523        let (ctx, store) = ctx_with_origin().await;
524        let del = CronDeleteTool::new(store);
525        let err = del
526            .call(&ctx, json!({"id": "nope"}))
527            .await
528            .unwrap_err()
529            .to_string();
530        assert!(err.contains("not found"), "got: {err}");
531    }
532
533    #[tokio::test]
534    async fn create_missing_required_args() {
535        let (ctx, store) = ctx_with_origin().await;
536        let tool = CronCreateTool::new(store);
537        let err1 = tool
538            .call(&ctx, json!({"prompt": "x"}))
539            .await
540            .unwrap_err()
541            .to_string();
542        assert!(err1.contains("requires `cron`"));
543        let err2 = tool
544            .call(&ctx, json!({"cron": "*/5 * * * *"}))
545            .await
546            .unwrap_err()
547            .to_string();
548        assert!(err2.contains("requires `prompt`"));
549    }
550
551    #[tokio::test]
552    async fn pause_then_resume_round_trip() {
553        let (ctx, store) = ctx_with_origin().await;
554        let create = CronCreateTool::new(store.clone());
555        let res = create
556            .call(&ctx, json!({"cron": "*/5 * * * *", "prompt": "x"}))
557            .await
558            .unwrap();
559        let id = res["id"].as_str().unwrap().to_string();
560
561        let pause = CronPauseTool::new(store.clone());
562        let res = pause.call(&ctx, json!({"id": id.clone()})).await.unwrap();
563        assert_eq!(res["paused"], true);
564        assert!(store.get(&id).await.unwrap().paused);
565
566        let resume = CronResumeTool::new(store.clone());
567        let res = resume.call(&ctx, json!({"id": id.clone()})).await.unwrap();
568        assert_eq!(res["paused"], false);
569        assert!(!store.get(&id).await.unwrap().paused);
570    }
571
572    #[tokio::test]
573    async fn pause_unknown_id_errors() {
574        let (ctx, store) = ctx_with_origin().await;
575        let pause = CronPauseTool::new(store);
576        let err = pause
577            .call(&ctx, json!({"id": "nope"}))
578            .await
579            .unwrap_err()
580            .to_string();
581        assert!(err.contains("not found"));
582    }
583
584    #[tokio::test]
585    async fn fallback_binding_id_uses_agent_id_without_inbound_origin() {
586        let cfg = AgentConfig {
587            id: "agent-z".into(),
588            model: ModelConfig {
589                provider: "x".into(),
590                model: "y".into(),
591            },
592            plugins: Vec::new(),
593            heartbeat: HeartbeatConfig::default(),
594            config: AgentRuntimeConfig::default(),
595            system_prompt: String::new(),
596            workspace: String::new(),
597            skills: Vec::new(),
598            skills_dir: "./skills".into(),
599            skill_overrides: Default::default(),
600            transcripts_dir: String::new(),
601            dreaming: DreamingYamlConfig::default(),
602            workspace_git: WorkspaceGitConfig::default(),
603            tool_rate_limits: None,
604            tool_args_validation: None,
605            extra_docs: Vec::new(),
606            inbound_bindings: Vec::new(),
607            allowed_tools: Vec::new(),
608            sender_rate_limit: None,
609            allowed_delegates: Vec::new(),
610            accept_delegates_from: Vec::new(),
611            description: String::new(),
612            google_auth: None,
613            credentials: Default::default(),
614            link_understanding: serde_json::Value::Null,
615            web_search: serde_json::Value::Null,
616            pairing_policy: serde_json::Value::Null,
617            language: None,
618            locale_prompts: Default::default(),
619            outbound_allowlist: OutboundAllowlistConfig::default(),
620            context_optimization: None,
621            dispatch_policy: Default::default(),
622            plan_mode: Default::default(),
623            remote_triggers: Vec::new(),
624            lsp: nexo_config::types::lsp::LspPolicy::default(),
625            config_tool: nexo_config::types::config_tool::ConfigToolPolicy::default(),
626            team: nexo_config::types::team::TeamPolicy::default(),
627            proactive: Default::default(),
628            repl: Default::default(),
629            auto_dream: None,
630            assistant_mode: None,
631            away_summary: None,
632            brief: None,
633            channels: None,
634            auto_approve: false,
635            extract_memories: None,
636            event_subscribers: Vec::new(),
637            tenant_id: None,
638            extensions_config: std::collections::BTreeMap::new(),
639            active: true,
640        };
641        let ctx = AgentContext::new(
642            "agent-z",
643            Arc::new(cfg),
644            AnyBroker::local(),
645            Arc::new(SessionManager::new(std::time::Duration::from_secs(60), 8)),
646        );
647        let store: Arc<dyn CronStore> = Arc::new(SqliteCronStore::open_memory().await.unwrap());
648        let tool = CronCreateTool::new(store.clone());
649        let res = tool
650            .call(&ctx, json!({"cron": "*/5 * * * *", "prompt": "x"}))
651            .await
652            .unwrap();
653        assert_eq!(res["binding_id"], "agent-z");
654    }
655}
nexo_core/agent/cron_tool.rs

nexo_core/agent/
cron_tool.rs
