| 1 | """Cron tool for scheduling reminders and tasks.""" |
| 2 | |
| 3 | from contextvars import ContextVar |
| 4 | from datetime import datetime |
| 5 | from typing import Any |
| 6 | |
| 7 | from nanobot.agent.tools.base import Tool, tool_parameters |
| 8 | from nanobot.agent.tools.schema import ( |
| 9 | BooleanSchema, |
| 10 | IntegerSchema, |
| 11 | StringSchema, |
| 12 | tool_parameters_schema, |
| 13 | ) |
| 14 | from nanobot.cron.service import CronService |
| 15 | from nanobot.cron.types import CronJob, CronJobState, CronSchedule |
| 16 | |
| 17 | _CRON_PARAMETERS = tool_parameters_schema( |
| 18 | action=StringSchema("Action to perform", enum=["add", "list", "remove"]), |
| 19 | name=StringSchema( |
| 20 | "Optional short human-readable label for the job " |
| 21 | "(e.g., 'weather-monitor', 'daily-standup'). Defaults to first 30 chars of message." |
| 22 | ), |
| 23 | message=StringSchema( |
| 24 | "REQUIRED when action='add'. Instruction for the agent to execute when the job triggers " |
| 25 | "(e.g., 'Send a reminder to WeChat: xxx' or 'Check system status and report'). " |
| 26 | "Not used for action='list' or action='remove'." |
| 27 | ), |
| 28 | every_seconds=IntegerSchema(0, description="Interval in seconds (for recurring tasks)"), |
| 29 | cron_expr=StringSchema("Cron expression like '0 9 * * *' (for scheduled tasks)"), |
| 30 | tz=StringSchema( |
| 31 | "Optional IANA timezone for cron expressions (e.g. 'America/Vancouver'). " |
| 32 | "When omitted with cron_expr, the tool's default timezone applies." |
| 33 | ), |
| 34 | at=StringSchema( |
| 35 | "ISO datetime for one-time execution (e.g. '2026-02-12T10:30:00'). " |
| 36 | "Naive values use the tool's default timezone." |
| 37 | ), |
| 38 | deliver=BooleanSchema( |
| 39 | description="Whether to deliver the execution result to the user channel (default true)", |
| 40 | default=True, |
| 41 | ), |
| 42 | job_id=StringSchema("REQUIRED when action='remove'. Job ID to remove (obtain via action='list')."), |
| 43 | required=["action"], |
| 44 | description=( |
| 45 | "Action-specific parameters: add requires a non-empty message plus one schedule " |
| 46 | "(every_seconds, cron_expr, or at); remove requires job_id; list only needs action. " |
| 47 | "Per-action requirements are enforced at runtime (see field descriptions) so the " |
| 48 | "top-level schema stays compatible with providers (e.g. OpenAI Codex/Responses) that " |
| 49 | "reject oneOf/anyOf/allOf/enum/not at the root of function parameters." |
| 50 | ), |
| 51 | ) |
| 52 | |
| 53 | |
| 54 | @tool_parameters(_CRON_PARAMETERS) |
| 55 | class CronTool(Tool): |
| 56 | """Tool to schedule reminders and recurring tasks.""" |
| 57 | |
| 58 | def __init__(self, cron_service: CronService, default_timezone: str = "UTC"): |
| 59 | self._cron = cron_service |
| 60 | self._default_timezone = default_timezone |
| 61 | self._channel: ContextVar[str] = ContextVar("cron_channel", default="") |
| 62 | self._chat_id: ContextVar[str] = ContextVar("cron_chat_id", default="") |
| 63 | self._in_cron_context: ContextVar[bool] = ContextVar("cron_in_context", default=False) |
| 64 | |
| 65 | def set_context(self, channel: str, chat_id: str) -> None: |
| 66 | """Set the current session context for delivery.""" |
| 67 | self._channel.set(channel) |
| 68 | self._chat_id.set(chat_id) |
| 69 | |
| 70 | def set_cron_context(self, active: bool): |
| 71 | """Mark whether the tool is executing inside a cron job callback.""" |
| 72 | return self._in_cron_context.set(active) |
| 73 | |
| 74 | def reset_cron_context(self, token) -> None: |
| 75 | """Restore previous cron context.""" |
| 76 | self._in_cron_context.reset(token) |
| 77 | |
| 78 | @staticmethod |
| 79 | def _validate_timezone(tz: str) -> str | None: |
| 80 | from zoneinfo import ZoneInfo |
| 81 | |
| 82 | try: |
| 83 | ZoneInfo(tz) |
| 84 | except (KeyError, Exception): |
| 85 | return f"Error: unknown timezone '{tz}'" |
| 86 | return None |
| 87 | |
| 88 | def _display_timezone(self, schedule: CronSchedule) -> str: |
| 89 | """Pick the most human-meaningful timezone for display.""" |
| 90 | return schedule.tz or self._default_timezone |
| 91 | |
| 92 | @staticmethod |
| 93 | def _format_timestamp(ms: int, tz_name: str) -> str: |
| 94 | from zoneinfo import ZoneInfo |
| 95 | |
| 96 | dt = datetime.fromtimestamp(ms / 1000, tz=ZoneInfo(tz_name)) |
| 97 | return f"{dt.isoformat()} ({tz_name})" |
| 98 | |
| 99 | @property |
| 100 | def name(self) -> str: |
| 101 | return "cron" |
| 102 | |
| 103 | @property |
| 104 | def description(self) -> str: |
| 105 | return ( |
| 106 | "Schedule reminders and recurring tasks. Actions: add, list, remove. " |
| 107 | f"If tz is omitted, cron expressions and naive ISO times default to {self._default_timezone}." |
| 108 | ) |
| 109 | |
| 110 | def validate_params(self, params: dict[str, Any]) -> list[str]: |
| 111 | errors = super().validate_params(params) |
| 112 | action = params.get("action") |
| 113 | if action == "add" and not str(params.get("message") or "").strip(): |
| 114 | errors.append("message is required when action='add'") |
| 115 | if action == "remove" and not str(params.get("job_id") or "").strip(): |
| 116 | errors.append("job_id is required when action='remove'") |
| 117 | return errors |
| 118 | |
| 119 | async def execute( |
| 120 | self, |
| 121 | action: str, |
| 122 | name: str | None = None, |
| 123 | message: str = "", |
| 124 | every_seconds: int | None = None, |
| 125 | cron_expr: str | None = None, |
| 126 | tz: str | None = None, |
| 127 | at: str | None = None, |
| 128 | job_id: str | None = None, |
| 129 | deliver: bool = True, |
| 130 | **kwargs: Any, |
| 131 | ) -> str: |
| 132 | if action == "add": |
| 133 | if self._in_cron_context.get(): |
| 134 | return "Error: cannot schedule new jobs from within a cron job execution" |
| 135 | return self._add_job(name, message, every_seconds, cron_expr, tz, at, deliver) |
| 136 | elif action == "list": |
| 137 | return self._list_jobs() |
| 138 | elif action == "remove": |
| 139 | return self._remove_job(job_id) |
| 140 | return f"Unknown action: {action}" |
| 141 | |
| 142 | def _add_job( |
| 143 | self, |
| 144 | name: str | None, |
| 145 | message: str, |
| 146 | every_seconds: int | None, |
| 147 | cron_expr: str | None, |
| 148 | tz: str | None, |
| 149 | at: str | None, |
| 150 | deliver: bool = True, |
| 151 | ) -> str: |
| 152 | if not message: |
| 153 | return ( |
| 154 | "Error: cron action='add' requires a non-empty 'message' parameter " |
| 155 | "describing what to do when the job triggers " |
| 156 | "(e.g. the reminder text). Retry including message=\"...\"." |
| 157 | ) |
| 158 | channel = self._channel.get() |
| 159 | chat_id = self._chat_id.get() |
| 160 | if not channel or not chat_id: |
| 161 | return "Error: no session context (channel/chat_id)" |
| 162 | if tz and not cron_expr: |
| 163 | return "Error: tz can only be used with cron_expr" |
| 164 | if tz: |
| 165 | if err := self._validate_timezone(tz): |
| 166 | return err |
| 167 | |
| 168 | # Build schedule |
| 169 | delete_after = False |
| 170 | if every_seconds: |
| 171 | schedule = CronSchedule(kind="every", every_ms=every_seconds * 1000) |
| 172 | elif cron_expr: |
| 173 | effective_tz = tz or self._default_timezone |
| 174 | if err := self._validate_timezone(effective_tz): |
| 175 | return err |
| 176 | schedule = CronSchedule(kind="cron", expr=cron_expr, tz=effective_tz) |
| 177 | elif at: |
| 178 | from zoneinfo import ZoneInfo |
| 179 | |
| 180 | try: |
| 181 | dt = datetime.fromisoformat(at) |
| 182 | except ValueError: |
| 183 | return f"Error: invalid ISO datetime format '{at}'. Expected format: YYYY-MM-DDTHH:MM:SS" |
| 184 | if dt.tzinfo is None: |
| 185 | if err := self._validate_timezone(self._default_timezone): |
| 186 | return err |
| 187 | dt = dt.replace(tzinfo=ZoneInfo(self._default_timezone)) |
| 188 | at_ms = int(dt.timestamp() * 1000) |
| 189 | schedule = CronSchedule(kind="at", at_ms=at_ms) |
| 190 | delete_after = True |
| 191 | else: |
| 192 | return "Error: either every_seconds, cron_expr, or at is required" |
| 193 | |
| 194 | job = self._cron.add_job( |
| 195 | name=name or message[:30], |
| 196 | schedule=schedule, |
| 197 | message=message, |
| 198 | deliver=deliver, |
| 199 | channel=channel, |
| 200 | to=chat_id, |
| 201 | delete_after_run=delete_after, |
| 202 | ) |
| 203 | return f"Created job '{job.name}' (id: {job.id})" |
| 204 | |
| 205 | def _format_timing(self, schedule: CronSchedule) -> str: |
| 206 | """Format schedule as a human-readable timing string.""" |
| 207 | if schedule.kind == "cron": |
| 208 | tz = f" ({schedule.tz})" if schedule.tz else "" |
| 209 | return f"cron: {schedule.expr}{tz}" |
| 210 | if schedule.kind == "every" and schedule.every_ms: |
| 211 | ms = schedule.every_ms |
| 212 | if ms % 3_600_000 == 0: |
| 213 | return f"every {ms // 3_600_000}h" |
| 214 | if ms % 60_000 == 0: |
| 215 | return f"every {ms // 60_000}m" |
| 216 | if ms % 1000 == 0: |
| 217 | return f"every {ms // 1000}s" |
| 218 | return f"every {ms}ms" |
| 219 | if schedule.kind == "at" and schedule.at_ms: |
| 220 | return f"at {self._format_timestamp(schedule.at_ms, self._display_timezone(schedule))}" |
| 221 | return schedule.kind |
| 222 | |
| 223 | def _format_state(self, state: CronJobState, schedule: CronSchedule) -> list[str]: |
| 224 | """Format job run state as display lines.""" |
| 225 | lines: list[str] = [] |
| 226 | display_tz = self._display_timezone(schedule) |
| 227 | if state.last_run_at_ms: |
| 228 | info = ( |
| 229 | f" Last run: {self._format_timestamp(state.last_run_at_ms, display_tz)}" |
| 230 | f" — {state.last_status or 'unknown'}" |
| 231 | ) |
| 232 | if state.last_error: |
| 233 | info += f" ({state.last_error})" |
| 234 | lines.append(info) |
| 235 | if state.next_run_at_ms: |
| 236 | lines.append(f" Next run: {self._format_timestamp(state.next_run_at_ms, display_tz)}") |
| 237 | return lines |
| 238 | |
| 239 | @staticmethod |
| 240 | def _system_job_purpose(job: CronJob) -> str: |
| 241 | if job.name == "dream": |
| 242 | return "Dream memory consolidation for long-term memory." |
| 243 | return "System-managed internal job." |
| 244 | |
| 245 | def _list_jobs(self) -> str: |
| 246 | jobs = self._cron.list_jobs() |
| 247 | if not jobs: |
| 248 | return "No scheduled jobs." |
| 249 | lines = [] |
| 250 | for j in jobs: |
| 251 | timing = self._format_timing(j.schedule) |
| 252 | parts = [f"- {j.name} (id: {j.id}, {timing})"] |
| 253 | if j.payload.kind == "system_event": |
| 254 | parts.append(f" Purpose: {self._system_job_purpose(j)}") |
| 255 | parts.append(" Protected: visible for inspection, but cannot be removed.") |
| 256 | parts.extend(self._format_state(j.state, j.schedule)) |
| 257 | lines.append("\n".join(parts)) |
| 258 | return "Scheduled jobs:\n" + "\n".join(lines) |
| 259 | |
| 260 | def _remove_job(self, job_id: str | None) -> str: |
| 261 | if not job_id: |
| 262 | return "Error: job_id is required for remove" |
| 263 | result = self._cron.remove_job(job_id) |
| 264 | if result == "removed": |
| 265 | return f"Removed job {job_id}" |
| 266 | if result == "protected": |
| 267 | job = self._cron.get_job(job_id) |
| 268 | if job and job.name == "dream": |
| 269 | return ( |
| 270 | "Cannot remove job `dream`.\n" |
| 271 | "This is a system-managed Dream memory consolidation job for long-term memory.\n" |
| 272 | "It remains visible so you can inspect it, but it cannot be removed." |
| 273 | ) |
| 274 | return ( |
| 275 | f"Cannot remove job `{job_id}`.\n" |
| 276 | "This is a protected system-managed cron job." |
| 277 | ) |
| 278 | return f"Job {job_id} not found" |
| 279 |