diff --git a/nanobot/agent/tools/cron.py b/nanobot/agent/tools/cron.py index 078c8ed..4b34ebc 100644 --- a/nanobot/agent/tools/cron.py +++ b/nanobot/agent/tools/cron.py @@ -1,11 +1,12 @@ """Cron tool for scheduling reminders and tasks.""" from contextvars import ContextVar +from datetime import datetime, timezone from typing import Any from nanobot.agent.tools.base import Tool from nanobot.cron.service import CronService -from nanobot.cron.types import CronSchedule +from nanobot.cron.types import CronJobState, CronSchedule class CronTool(Tool): @@ -143,49 +144,49 @@ class CronTool(Tool): ) return f"Created job '{job.name}' (id: {job.id})" + @staticmethod + def _format_timing(schedule: CronSchedule) -> str: + """Format schedule as a human-readable timing string.""" + if schedule.kind == "cron": + tz = f" ({schedule.tz})" if schedule.tz else "" + return f"cron: {schedule.expr}{tz}" + if schedule.kind == "every" and schedule.every_ms: + secs = schedule.every_ms // 1000 + if secs >= 3600: + return f"every {secs // 3600}h" + if secs >= 60: + return f"every {secs // 60}m" + return f"every {secs}s" + if schedule.kind == "at" and schedule.at_ms: + dt = datetime.fromtimestamp(schedule.at_ms / 1000, tz=timezone.utc) + return f"at {dt.isoformat()}" + return schedule.kind + + @staticmethod + def _format_state(state: CronJobState) -> list[str]: + """Format job run state as display lines.""" + lines: list[str] = [] + if state.last_run_at_ms: + last_dt = datetime.fromtimestamp(state.last_run_at_ms / 1000, tz=timezone.utc) + info = f" Last run: {last_dt.isoformat()} — {state.last_status or 'unknown'}" + if state.last_error: + info += f" ({state.last_error})" + lines.append(info) + if state.next_run_at_ms: + next_dt = datetime.fromtimestamp(state.next_run_at_ms / 1000, tz=timezone.utc) + lines.append(f" Next run: {next_dt.isoformat()}") + return lines + def _list_jobs(self) -> str: jobs = self._cron.list_jobs() if not jobs: return "No scheduled jobs." lines = [] for j in jobs: - s = j.schedule - if s.kind == "cron": - timing = f"cron: {s.expr}" - if s.tz: - timing += f" ({s.tz})" - elif s.kind == "every" and s.every_ms: - secs = s.every_ms // 1000 - if secs >= 3600: - timing = f"every {secs // 3600}h" - elif secs >= 60: - timing = f"every {secs // 60}m" - else: - timing = f"every {secs}s" - elif s.kind == "at" and s.at_ms: - from datetime import datetime, timezone - - dt = datetime.fromtimestamp(s.at_ms / 1000, tz=timezone.utc) - timing = f"at {dt.isoformat()}" - else: - timing = s.kind + timing = self._format_timing(j.schedule) status = "enabled" if j.enabled else "disabled" parts = [f"- {j.name} (id: {j.id}, {timing}, {status})"] - if j.state.last_run_at_ms: - from datetime import datetime, timezone - - last_dt = datetime.fromtimestamp(j.state.last_run_at_ms / 1000, tz=timezone.utc) - last_info = ( - f" Last run: {last_dt.isoformat()} — {j.state.last_status or 'unknown'}" - ) - if j.state.last_error: - last_info += f" ({j.state.last_error})" - parts.append(last_info) - if j.state.next_run_at_ms: - from datetime import datetime, timezone - - next_dt = datetime.fromtimestamp(j.state.next_run_at_ms / 1000, tz=timezone.utc) - parts.append(f" Next run: {next_dt.isoformat()}") + parts.extend(self._format_state(j.state)) lines.append("\n".join(parts)) return "Scheduled jobs:\n" + "\n".join(lines)