From 9106b783e88f3c4a96ab51d27d59f86af887a2ed Mon Sep 17 00:00:00 2001 From: Travis Ha Date: Tue, 26 May 2026 19:39:07 -0700 Subject: [PATCH 1/2] feat(schedule): add gaia schedule CLI + cron-based dispatch (#892) GAIA had no way to run a skill or prompt on a recurring schedule, so every recurring use case (morning brief, hourly watcher, weekly digest) needed an external cron plus manual wiring. This adds a `gaia schedule` command group backed by a hand-editable ~/.gaia/schedules.toml store and an APScheduler daemon: register a prompt with a cron expression and the agent runs it on schedule in a fresh session, delivering output to a configurable sink (stdout, file, desktop notification, or Telegram). The --prompt path is fully wired; --skill raises an actionable error pending the skill format (#888). Sinks fail loudly on delivery errors per the no-fallback rule. Tests, CLI docs, and the optional OS service installers are follow-ups, not included here. --- setup.py | 6 ++ src/gaia/cli.py | 144 ++++++++++++++++++++++++++++++++++ src/gaia/schedule/__init__.py | 11 +++ src/gaia/schedule/daemon.py | 79 +++++++++++++++++++ src/gaia/schedule/runner.py | 45 +++++++++++ src/gaia/schedule/sinks.py | 128 ++++++++++++++++++++++++++++++ src/gaia/schedule/store.py | 140 +++++++++++++++++++++++++++++++++ 7 files changed, 553 insertions(+) create mode 100644 src/gaia/schedule/__init__.py create mode 100644 src/gaia/schedule/daemon.py create mode 100644 src/gaia/schedule/runner.py create mode 100644 src/gaia/schedule/sinks.py create mode 100644 src/gaia/schedule/store.py diff --git a/setup.py b/setup.py index bc147d506..dd7334b1e 100644 --- a/setup.py +++ b/setup.py @@ -28,6 +28,7 @@ "gaia.llm.providers", "gaia.audio", "gaia.chat", + "gaia.schedule", "gaia.ui", "gaia.ui.routers", "gaia.database", @@ -127,6 +128,11 @@ "beautifulsoup4", "watchdog>=2.1.0", "pillow>=9.0.0", + # Cron-based scheduler (issue #892): apscheduler drives the daemon; + # tomli/tomli-w read+write ~/.gaia/schedules.toml (tomllib is stdlib on 3.11+). + "apscheduler>=3.10.0", + "tomli-w>=1.0.0", + "tomli>=2.0.0; python_version < '3.11'", ], extras_require={ "image": [ diff --git a/src/gaia/cli.py b/src/gaia/cli.py index f95fdda4d..1be5dd1b6 100644 --- a/src/gaia/cli.py +++ b/src/gaia/cli.py @@ -1773,6 +1773,69 @@ def build_parser(): telegram_parser.set_defaults(action="telegram") + # Schedule command — cron-based recurring skill/prompt dispatch (issue #892) + schedule_parser = subparsers.add_parser( + "schedule", + help="Run a skill or prompt on a cron schedule " + "(add|list|show|remove|pause|resume|run|daemon)", + parents=[parent_parser], + ) + schedule_subparsers = schedule_parser.add_subparsers( + dest="schedule_action", help="schedule action to perform" + ) + + s_add = schedule_subparsers.add_parser("add", help="Register a new schedule") + s_add.add_argument("--name", required=True, help="Unique schedule name") + s_add.add_argument( + "--cron", required=True, help='Cron expression, e.g. "0 7 * * 1-5"' + ) + s_add_what = s_add.add_mutually_exclusive_group(required=True) + s_add_what.add_argument("--prompt", help="One-shot prompt to run on each fire") + s_add_what.add_argument( + "--skill", help="Skill to run (pending skill-format support, #888)" + ) + s_add.add_argument( + "--sink", + default="stdout", + help="Where output goes: stdout (default) | file: | notification | telegram", + ) + s_add.add_argument( + "--to", help="Recipient for the telegram sink (Telegram user/chat id)" + ) + + s_list = schedule_subparsers.add_parser( + "list", help="List schedules with next fire time" + ) + s_list.set_defaults(schedule_action="list") + + s_show = schedule_subparsers.add_parser("show", help="Show one schedule") + s_show.add_argument("name", help="Schedule name") + + s_remove = schedule_subparsers.add_parser("remove", help="Remove a schedule") + s_remove.add_argument("name", help="Schedule name") + + s_pause = schedule_subparsers.add_parser( + "pause", help="Disable a schedule without deleting it" + ) + s_pause.add_argument("name", help="Schedule name") + + s_resume = schedule_subparsers.add_parser( + "resume", help="Re-enable a paused schedule" + ) + s_resume.add_argument("name", help="Schedule name") + + s_run = schedule_subparsers.add_parser( + "run", help="Fire a schedule once now (for testing)" + ) + s_run.add_argument("name", help="Schedule name") + + s_daemon = schedule_subparsers.add_parser( + "daemon", help="Run the long-lived scheduler (blocks until interrupted)" + ) + s_daemon.set_defaults(schedule_action="daemon") + + schedule_parser.set_defaults(action="schedule") + # Add model download command download_parser = subparsers.add_parser( "download", @@ -2619,6 +2682,83 @@ def build_parser(): return parser +def _handle_schedule(args): + """Dispatch `gaia schedule ` (issue #892).""" + from gaia.schedule import daemon as schedule_daemon + from gaia.schedule import runner as schedule_runner + from gaia.schedule.store import Schedule, ScheduleStore + + action = getattr(args, "schedule_action", None) + store = ScheduleStore() + + if action == "add": + sink_args = {} + if getattr(args, "to", None): + sink_args["to"] = args.to + schedule = Schedule( + name=args.name, + cron=args.cron, + skill=getattr(args, "skill", None), + prompt=getattr(args, "prompt", None), + sink=args.sink, + sink_args=sink_args, + ) + store.add(schedule) + print(f"✅ Added schedule {schedule.name!r} ({schedule.cron})") + return + + if action == "list": + schedules = store.load() + if not schedules: + print("No schedules registered. Add one with `gaia schedule add`.") + return + for s in schedules.values(): + state = "enabled" if s.enabled else "paused" + nxt = schedule_daemon.next_fire_time(s.cron) if s.enabled else None + print( + f"{s.name} [{state}] cron={s.cron!r} sink={s.sink}" + + (f" next={nxt}" if nxt else "") + ) + return + + if action == "show": + s = store.get(args.name) + print(f"name: {s.name}") + print(f"cron: {s.cron}") + print(f"target: {'skill=' + s.skill if s.skill else 'prompt=' + s.prompt}") + print(f"sink: {s.sink} args={s.sink_args}") + print(f"enabled: {s.enabled}") + print(f"created_at: {s.created_at}") + print(f"next_fire: {schedule_daemon.next_fire_time(s.cron)}") + return + + if action == "remove": + store.remove(args.name) + print(f"✅ Removed schedule {args.name!r}") + return + + if action in ("pause", "resume"): + enabled = action == "resume" + store.set_enabled(args.name, enabled) + print(f"✅ {'Resumed' if enabled else 'Paused'} schedule {args.name!r}") + return + + if action == "run": + schedule = store.get(args.name) + schedule_runner.fire(schedule) + return + + if action == "daemon": + schedule_daemon.run_daemon() + return + + print( + "No schedule action specified. Use: " + "gaia schedule add|list|show|remove|pause|resume|run|daemon", + file=sys.stderr, + ) + + def main(): parser = build_parser() log = get_logger(__name__) @@ -2768,6 +2908,10 @@ def main(): ) return + if args.action == "schedule": + _handle_schedule(args) + return + # Handle core Gaia CLI commands if args.action in ["prompt", "chat", "browse", "analyze", "talk", "stats"]: kwargs = { diff --git a/src/gaia/schedule/__init__.py b/src/gaia/schedule/__init__.py new file mode 100644 index 000000000..b60e1a6a9 --- /dev/null +++ b/src/gaia/schedule/__init__.py @@ -0,0 +1,11 @@ +"""Cron-based recurring execution for GAIA skills and prompts (issue #892). + +A minimal scheduler: register a skill or prompt to run on a cron schedule, and +deliver the agent's output to a configurable sink (stdout, file, notification, +Telegram). Persisted in ``~/.gaia/schedules.toml`` and driven by a long-running +``gaia schedule daemon``. +""" + +from gaia.schedule.store import Schedule, ScheduleStore + +__all__ = ["Schedule", "ScheduleStore"] diff --git a/src/gaia/schedule/daemon.py b/src/gaia/schedule/daemon.py new file mode 100644 index 000000000..6ced59c05 --- /dev/null +++ b/src/gaia/schedule/daemon.py @@ -0,0 +1,79 @@ +"""Long-running scheduler daemon. + +Reads the schedule store, arms an APScheduler cron trigger per enabled schedule, +and blocks until interrupted. Each trigger fires :func:`runner.fire`. +""" + +from __future__ import annotations + +import signal +import threading +from pathlib import Path +from typing import Optional + +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger + +from gaia.logger import get_logger +from gaia.schedule import runner +from gaia.schedule.store import DEFAULT_STORE_PATH, Schedule, ScheduleStore + +log = get_logger(__name__) + + +def _job(schedule: Schedule) -> None: + # A failing job must not kill the daemon, but it must be loud (no silent + # swallow): log with full traceback and keep the other schedules alive. + try: + runner.fire(schedule) + except Exception: + log.exception("schedule %r failed", schedule.name) + + +def build_scheduler(store: ScheduleStore) -> BackgroundScheduler: + """Create a scheduler with one cron job per enabled schedule.""" + scheduler = BackgroundScheduler() + schedules = store.load() + armed = 0 + for schedule in schedules.values(): + if not schedule.enabled: + log.info("skipping disabled schedule %r", schedule.name) + continue + scheduler.add_job( + _job, + trigger=CronTrigger.from_crontab(schedule.cron), + args=[schedule], + id=schedule.name, + name=schedule.name, + replace_existing=True, + ) + armed += 1 + log.info("armed %d schedule(s) from %s", armed, store.path) + return scheduler + + +def run_daemon(store_path: Path = DEFAULT_STORE_PATH) -> None: + """Start the scheduler and block until SIGINT/SIGTERM.""" + store = ScheduleStore(store_path) + scheduler = build_scheduler(store) + scheduler.start() + + stop = threading.Event() + for sig in (signal.SIGINT, signal.SIGTERM): + signal.signal(sig, lambda *_: stop.set()) + + log.info("schedule daemon running; press Ctrl-C to stop") + try: + stop.wait() + finally: + scheduler.shutdown(wait=False) + log.info("schedule daemon stopped") + + +def next_fire_time(cron: str) -> Optional[str]: + """Human-readable next fire time for a cron expression (for `list`).""" + from datetime import datetime + + trigger = CronTrigger.from_crontab(cron) + nxt = trigger.get_next_fire_time(None, datetime.now(trigger.timezone)) + return nxt.isoformat() if nxt else None diff --git a/src/gaia/schedule/runner.py b/src/gaia/schedule/runner.py new file mode 100644 index 000000000..48c9a7dbb --- /dev/null +++ b/src/gaia/schedule/runner.py @@ -0,0 +1,45 @@ +"""Execute a single scheduled job and deliver its output. + +Each fire runs in a fresh agent session (clean state, not a continuation) and +routes the result through the schedule's configured sink. +""" + +from __future__ import annotations + +from gaia.logger import get_logger +from gaia.schedule import sinks +from gaia.schedule.store import Schedule + +log = get_logger(__name__) + + +def resolve_input(schedule: Schedule) -> str: + """Return the prompt text to send to the agent for this schedule. + + ``--skill`` resolution depends on the agentskills.io skill format (#888), + which has not landed yet; fail loudly rather than guess. + """ + if schedule.prompt: + return schedule.prompt + raise NotImplementedError( + f"schedule {schedule.name!r} uses --skill {schedule.skill!r}, but skill-format " + f"resolution is not available yet (blocked on #888). Use --prompt for now." + ) + + +def fire(schedule: Schedule) -> str: + """Run one scheduled job: fresh agent session -> sink. Returns the output.""" + # Imported lazily so the scheduler/store can be used without spinning up the + # full agent stack (and so tests can register schedules without an LLM). + from gaia.chat.sdk import AgentConfig, AgentSDK + + prompt = resolve_input(schedule) + log.info("schedule %r firing (sink=%s)", schedule.name, schedule.sink) + + sdk = AgentSDK(AgentConfig()) + response = sdk.send(prompt, no_history=True) + output = response.text + + sinks.dispatch(schedule.sink, schedule.sink_args, output) + log.info("schedule %r delivered to sink %s", schedule.name, schedule.sink) + return output diff --git a/src/gaia/schedule/sinks.py b/src/gaia/schedule/sinks.py new file mode 100644 index 000000000..d0d3773c5 --- /dev/null +++ b/src/gaia/schedule/sinks.py @@ -0,0 +1,128 @@ +"""Output sinks for scheduled runs. + +A sink decides *where* a scheduled run's output goes. Per the GAIA no-fallback +rule, a sink that cannot deliver raises an actionable error — it never silently +swallows the failure. +""" + +from __future__ import annotations + +import os +import platform +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict + +import requests + +TELEGRAM_API = "https://api.telegram.org" + + +def dispatch(sink: str, sink_args: Dict[str, Any], output: str) -> None: + """Route ``output`` to the named sink. + + Raises on any delivery failure with a message naming what failed, what the + caller should do, and where to look. + """ + if sink == "stdout": + _to_stdout(output) + elif sink == "notification": + _to_notification(output, sink_args) + elif sink == "telegram": + _to_telegram(output, sink_args) + elif sink.startswith("file:"): + _to_file(sink[len("file:") :], output) + elif sink == "file": + path = sink_args.get("path") + if not path: + raise ValueError( + "file sink requires a path: use --sink file:/path/to/log.md " + "or pass sink_args.path in schedules.toml" + ) + _to_file(path, output) + else: + raise ValueError( + f"unknown sink {sink!r}; valid sinks are: stdout, file:, " + f"notification, telegram" + ) + + +def _to_stdout(output: str) -> None: + print(output, flush=True) + + +def _to_file(path: str, output: str) -> None: + target = Path(os.path.expanduser(path)) + try: + target.parent.mkdir(parents=True, exist_ok=True) + with open(target, "a", encoding="utf-8") as f: + f.write(output.rstrip("\n") + "\n") + except OSError as e: + raise OSError( + f"file sink could not append to {target}: {e}. " + f"Check the path exists and is writable." + ) from e + + +def _to_notification(output: str, sink_args: Dict[str, Any]) -> None: + title = sink_args.get("title", "GAIA schedule") + system = platform.system() + try: + if system == "Darwin": + script = f'display notification {_osa(output)} with title {_osa(title)}' + subprocess.run(["osascript", "-e", script], check=True) + elif system == "Linux": + subprocess.run(["notify-send", title, output], check=True) + else: + raise NotImplementedError( + f"notification sink is not implemented for {system!r}; " + f"use --sink stdout, file:, or telegram instead" + ) + except FileNotFoundError as e: + tool = "osascript" if system == "Darwin" else "notify-send" + raise FileNotFoundError( + f"notification sink needs {tool!r} on PATH ({system}): {e}. " + f"Install it or choose a different --sink." + ) from e + except subprocess.CalledProcessError as e: + raise RuntimeError( + f"notification sink failed to post (exit {e.returncode}). " + f"Choose a different --sink or check the desktop notification daemon." + ) from e + + +def _osa(text: str) -> str: + """Quote a string for embedding inside an AppleScript literal.""" + return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def _to_telegram(output: str, sink_args: Dict[str, Any]) -> None: + token = sink_args.get("token") or os.environ.get("GAIA_TELEGRAM_TOKEN") + to = sink_args.get("to") + if not token: + raise ValueError( + "telegram sink needs a bot token: set GAIA_TELEGRAM_TOKEN or pass " + "sink_args.token in schedules.toml" + ) + if not to: + raise ValueError( + "telegram sink needs a recipient: pass --to " + "(stored as sink_args.to)" + ) + url = f"{TELEGRAM_API}/bot{token}/sendMessage" + try: + resp = requests.post( + url, json={"chat_id": to, "text": output}, timeout=30 + ) + except requests.RequestException as e: + raise RuntimeError( + f"telegram sink could not reach {TELEGRAM_API}: {e}. " + f"Check network connectivity." + ) from e + if resp.status_code != 200: + raise RuntimeError( + f"telegram sink got HTTP {resp.status_code} from sendMessage: " + f"{resp.text[:200]}. Verify the bot token and that chat_id {to!r} " + f"has started a conversation with the bot." + ) diff --git a/src/gaia/schedule/store.py b/src/gaia/schedule/store.py new file mode 100644 index 000000000..5de359b38 --- /dev/null +++ b/src/gaia/schedule/store.py @@ -0,0 +1,140 @@ +"""Schedule persistence: read/write ``~/.gaia/schedules.toml``. + +The store is the single source of truth so schedules survive daemon restarts. +The file is intentionally human-readable and hand-editable. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional + +# tomllib is stdlib on 3.11+; fall back to tomli on 3.10 (python_requires>=3.10). +try: + import tomllib # type: ignore[import-not-found] +except ModuleNotFoundError: # pragma: no cover - exercised only on 3.10 + import tomli as tomllib # type: ignore[no-redef] + +import tomli_w + +DEFAULT_STORE_PATH = Path(os.path.expanduser("~/.gaia/schedules.toml")) + + +@dataclass +class Schedule: + """One scheduled job. Exactly one of ``skill`` or ``prompt`` must be set.""" + + name: str + cron: str + skill: Optional[str] = None + prompt: Optional[str] = None + sink: str = "stdout" + sink_args: Dict[str, Any] = field(default_factory=dict) + enabled: bool = True + created_at: str = "" + + def __post_init__(self) -> None: + if bool(self.skill) == bool(self.prompt): + raise ValueError( + f"schedule {self.name!r}: set exactly one of --skill or --prompt " + f"(got skill={self.skill!r}, prompt={self.prompt!r})" + ) + if not self.created_at: + self.created_at = datetime.now(timezone.utc).isoformat() + + def to_toml_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "cron": self.cron, + "sink": self.sink, + "enabled": self.enabled, + "created_at": self.created_at, + } + if self.skill: + d["skill"] = self.skill + if self.prompt: + d["prompt"] = self.prompt + if self.sink_args: + d["sink_args"] = self.sink_args + return d + + @classmethod + def from_toml_dict(cls, name: str, data: Dict[str, Any]) -> "Schedule": + return cls( + name=name, + cron=data["cron"], + skill=data.get("skill"), + prompt=data.get("prompt"), + sink=data.get("sink", "stdout"), + sink_args=data.get("sink_args", {}) or {}, + enabled=data.get("enabled", True), + created_at=data.get("created_at", ""), + ) + + +class ScheduleStore: + """Load/save a collection of :class:`Schedule` objects to a TOML file.""" + + def __init__(self, path: Path = DEFAULT_STORE_PATH): + self.path = Path(path) + + def load(self) -> Dict[str, Schedule]: + if not self.path.exists(): + return {} + with open(self.path, "rb") as f: + raw = tomllib.load(f) + schedules = raw.get("schedules", {}) + return { + name: Schedule.from_toml_dict(name, data) + for name, data in schedules.items() + } + + def save(self, schedules: Dict[str, Schedule]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + doc = { + "schedules": {s.name: s.to_toml_dict() for s in schedules.values()} + } + with open(self.path, "wb") as f: + tomli_w.dump(doc, f) + + def add(self, schedule: Schedule) -> None: + schedules = self.load() + if schedule.name in schedules: + raise ValueError( + f"schedule {schedule.name!r} already exists in {self.path}; " + f"remove it first or pick a different --name" + ) + schedules[schedule.name] = schedule + self.save(schedules) + + def remove(self, name: str) -> None: + schedules = self.load() + if name not in schedules: + raise KeyError( + f"no schedule named {name!r} in {self.path}; " + f"run `gaia schedule list` to see registered schedules" + ) + del schedules[name] + self.save(schedules) + + def get(self, name: str) -> Schedule: + schedules = self.load() + if name not in schedules: + raise KeyError( + f"no schedule named {name!r} in {self.path}; " + f"run `gaia schedule list` to see registered schedules" + ) + return schedules[name] + + def set_enabled(self, name: str, enabled: bool) -> Schedule: + schedules = self.load() + if name not in schedules: + raise KeyError( + f"no schedule named {name!r} in {self.path}; " + f"run `gaia schedule list` to see registered schedules" + ) + schedules[name].enabled = enabled + self.save(schedules) + return schedules[name] From 0055a01a5f0d27497a3896a81360cc81a5eb0068 Mon Sep 17 00:00:00 2001 From: Travis Ha Date: Sun, 21 Jun 2026 09:27:31 -0700 Subject: [PATCH 2/2] feat(schedule): ScheduleStore protocol, run-state fields, tests + docs Prepare the schedule store to converge with the Agent UI's SQLite store (#1566) and close the test/doc gaps the initial CLI PR deferred. - Schedule gains last_run/next_run/session_ref so one dataclass serves both the hand-editable TOML layer and the future SQLite/run-history store; TOML serialization omits each field when unset. - ScheduleStore becomes a runtime_checkable Protocol (load/save/add/remove/get/set_enabled/mark_run) and the TOML implementation is renamed TomlScheduleStore, so a database-backed store can drop in without changing the daemon or CLI. - mark_run records last_run/next_run after a successful fire (daemon and `gaia schedule run`); a failed fire still logs loudly and leaves run-state untouched, keeping the daemon alive. - build_scheduler now depends only on the Protocol surface (no concrete store.path access) so the UI server can embed the same engine on a background thread instead of requiring a separate daemon process. - Add hermetic unit tests for the store, sinks, and daemon (LLM and network mocked), and document `gaia schedule` in docs/reference/cli.mdx. --- docs/reference/cli.mdx | 89 ++++++++ src/gaia/cli.py | 11 +- src/gaia/schedule/__init__.py | 4 +- src/gaia/schedule/daemon.py | 24 ++- src/gaia/schedule/sinks.py | 7 +- src/gaia/schedule/store.py | 61 +++++- tests/unit/test_schedule_daemon.py | 239 ++++++++++++++++++++++ tests/unit/test_schedule_sinks.py | 214 ++++++++++++++++++++ tests/unit/test_schedule_store.py | 312 +++++++++++++++++++++++++++++ 9 files changed, 941 insertions(+), 20 deletions(-) create mode 100644 tests/unit/test_schedule_daemon.py create mode 100644 tests/unit/test_schedule_sinks.py create mode 100644 tests/unit/test_schedule_store.py diff --git a/docs/reference/cli.mdx b/docs/reference/cli.mdx index 0f05d12bc..f63c4e0e9 100644 --- a/docs/reference/cli.mdx +++ b/docs/reference/cli.mdx @@ -1528,6 +1528,95 @@ Manage agent memory: run day-zero onboarding and view memory statistics. --- +## Schedule + +Run a prompt on a recurring cron schedule and route its output to a sink. Schedules are stored in `~/.gaia/schedules.toml` (hand-editable). The `daemon` action runs a long-lived scheduler that fires each enabled schedule when due. + +```bash +gaia schedule {add,list,show,remove,pause,resume,run,daemon} [OPTIONS] +``` + +### Subcommands + +| Action | Description | +|--------|-------------| +| `add` | Register a new schedule (requires `--name`, `--cron`, and exactly one of `--prompt`/`--skill`) | +| `list` | List all schedules with their next fire time | +| `show ` | Show one schedule's full configuration and next fire time | +| `remove ` | Delete a schedule | +| `pause ` | Disable a schedule without deleting it | +| `resume ` | Re-enable a paused schedule | +| `run ` | Fire a schedule once now (for testing) | +| `daemon` | Run the long-lived scheduler (blocks until interrupted) | + +### `add` Options + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `--name` | string | _required_ | Unique schedule name | +| `--cron` | string | _required_ | Cron expression, e.g. `"0 7 * * 1-5"` | +| `--prompt` | string | - | One-shot prompt to run on each fire (mutually exclusive with `--skill`) | +| `--skill` | string | - | Skill to run (mutually exclusive with `--prompt`) — see note below | +| `--sink` | string | stdout | Where output goes: `stdout` \| `file:` \| `notification` \| `telegram` | +| `--to` | string | - | Recipient for the `telegram` sink (Telegram user/chat id) | + +**Examples:** + + +```bash Add a Weekday Morning Brief +gaia schedule add \ + --name morning-brief \ + --cron "0 7 * * 1-5" \ + --prompt "Give me a 3-bullet summary of today's priorities" +``` + +```bash List Schedules +gaia schedule list +``` + +```bash Fire Once (Test) +gaia schedule run morning-brief +``` + +```bash Pause and Resume +gaia schedule pause morning-brief +gaia schedule resume morning-brief +``` + +```bash Run the Scheduler +gaia schedule daemon +``` + + +### Sinks + +A sink decides where each scheduled run's output goes. Set it with `--sink` on `add` (default `stdout`): + +| Sink | Description | +|------|-------------| +| `stdout` | Print output to the terminal (default) | +| `file:` | Append output to the given file (created if missing) | +| `notification` | Desktop notification (macOS `osascript` / Linux `notify-send`) | +| `telegram` | Send via a Telegram bot — needs a bot token (`GAIA_TELEGRAM_TOKEN` env var or `sink_args.token` in `schedules.toml`) and a recipient via `--to` | + +**Telegram example:** + +```bash +export GAIA_TELEGRAM_TOKEN="" +gaia schedule add \ + --name daily-digest \ + --cron "0 18 * * *" \ + --prompt "Summarize what changed today" \ + --sink telegram \ + --to 123456789 +``` + + +`--skill` is not yet implemented — running a skill-backed schedule raises an error pending the skill format ([#888](https://github.com/amd/gaia/issues/888)). Use `--prompt` for now. + + +--- + ## Utility Commands ### Stats Command diff --git a/src/gaia/cli.py b/src/gaia/cli.py index 1be5dd1b6..c4d2c64ac 100644 --- a/src/gaia/cli.py +++ b/src/gaia/cli.py @@ -2686,10 +2686,10 @@ def _handle_schedule(args): """Dispatch `gaia schedule ` (issue #892).""" from gaia.schedule import daemon as schedule_daemon from gaia.schedule import runner as schedule_runner - from gaia.schedule.store import Schedule, ScheduleStore + from gaia.schedule.store import Schedule, TomlScheduleStore action = getattr(args, "schedule_action", None) - store = ScheduleStore() + store = TomlScheduleStore() if action == "add": sink_args = {} @@ -2744,8 +2744,15 @@ def _handle_schedule(args): return if action == "run": + from datetime import datetime, timezone + schedule = store.get(args.name) schedule_runner.fire(schedule) + store.mark_run( + schedule.name, + datetime.now(timezone.utc).isoformat(), + next_run=schedule_daemon.next_fire_time(schedule.cron), + ) return if action == "daemon": diff --git a/src/gaia/schedule/__init__.py b/src/gaia/schedule/__init__.py index b60e1a6a9..319deb260 100644 --- a/src/gaia/schedule/__init__.py +++ b/src/gaia/schedule/__init__.py @@ -6,6 +6,6 @@ ``gaia schedule daemon``. """ -from gaia.schedule.store import Schedule, ScheduleStore +from gaia.schedule.store import Schedule, ScheduleStore, TomlScheduleStore -__all__ = ["Schedule", "ScheduleStore"] +__all__ = ["Schedule", "ScheduleStore", "TomlScheduleStore"] diff --git a/src/gaia/schedule/daemon.py b/src/gaia/schedule/daemon.py index 6ced59c05..647c71d96 100644 --- a/src/gaia/schedule/daemon.py +++ b/src/gaia/schedule/daemon.py @@ -8,6 +8,7 @@ import signal import threading +from datetime import datetime, timezone from pathlib import Path from typing import Optional @@ -16,18 +17,29 @@ from gaia.logger import get_logger from gaia.schedule import runner -from gaia.schedule.store import DEFAULT_STORE_PATH, Schedule, ScheduleStore +from gaia.schedule.store import ( + DEFAULT_STORE_PATH, + Schedule, + ScheduleStore, + TomlScheduleStore, +) log = get_logger(__name__) -def _job(schedule: Schedule) -> None: +def _job(schedule: Schedule, store: ScheduleStore) -> None: # A failing job must not kill the daemon, but it must be loud (no silent # swallow): log with full traceback and keep the other schedules alive. try: runner.fire(schedule) except Exception: log.exception("schedule %r failed", schedule.name) + return + store.mark_run( + schedule.name, + datetime.now(timezone.utc).isoformat(), + next_run=next_fire_time(schedule.cron), + ) def build_scheduler(store: ScheduleStore) -> BackgroundScheduler: @@ -42,19 +54,19 @@ def build_scheduler(store: ScheduleStore) -> BackgroundScheduler: scheduler.add_job( _job, trigger=CronTrigger.from_crontab(schedule.cron), - args=[schedule], + args=[schedule, store], id=schedule.name, name=schedule.name, replace_existing=True, ) armed += 1 - log.info("armed %d schedule(s) from %s", armed, store.path) + log.info("armed %d schedule(s)", armed) return scheduler def run_daemon(store_path: Path = DEFAULT_STORE_PATH) -> None: """Start the scheduler and block until SIGINT/SIGTERM.""" - store = ScheduleStore(store_path) + store = TomlScheduleStore(store_path) scheduler = build_scheduler(store) scheduler.start() @@ -62,7 +74,7 @@ def run_daemon(store_path: Path = DEFAULT_STORE_PATH) -> None: for sig in (signal.SIGINT, signal.SIGTERM): signal.signal(sig, lambda *_: stop.set()) - log.info("schedule daemon running; press Ctrl-C to stop") + log.info("schedule daemon running (store=%s); press Ctrl-C to stop", store.path) try: stop.wait() finally: diff --git a/src/gaia/schedule/sinks.py b/src/gaia/schedule/sinks.py index d0d3773c5..7c255bd9f 100644 --- a/src/gaia/schedule/sinks.py +++ b/src/gaia/schedule/sinks.py @@ -10,7 +10,6 @@ import os import platform import subprocess -import sys from pathlib import Path from typing import Any, Dict @@ -70,7 +69,7 @@ def _to_notification(output: str, sink_args: Dict[str, Any]) -> None: system = platform.system() try: if system == "Darwin": - script = f'display notification {_osa(output)} with title {_osa(title)}' + script = f"display notification {_osa(output)} with title {_osa(title)}" subprocess.run(["osascript", "-e", script], check=True) elif system == "Linux": subprocess.run(["notify-send", title, output], check=True) @@ -112,9 +111,7 @@ def _to_telegram(output: str, sink_args: Dict[str, Any]) -> None: ) url = f"{TELEGRAM_API}/bot{token}/sendMessage" try: - resp = requests.post( - url, json={"chat_id": to, "text": output}, timeout=30 - ) + resp = requests.post(url, json={"chat_id": to, "text": output}, timeout=30) except requests.RequestException as e: raise RuntimeError( f"telegram sink could not reach {TELEGRAM_API}: {e}. " diff --git a/src/gaia/schedule/store.py b/src/gaia/schedule/store.py index 5de359b38..4e8983d98 100644 --- a/src/gaia/schedule/store.py +++ b/src/gaia/schedule/store.py @@ -10,7 +10,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Protocol, runtime_checkable # tomllib is stdlib on 3.11+; fall back to tomli on 3.10 (python_requires>=3.10). try: @@ -34,6 +34,9 @@ class Schedule: sink: str = "stdout" sink_args: Dict[str, Any] = field(default_factory=dict) enabled: bool = True + last_run: Optional[str] = None + next_run: Optional[str] = None + session_ref: Optional[str] = None created_at: str = "" def __post_init__(self) -> None: @@ -58,6 +61,12 @@ def to_toml_dict(self) -> Dict[str, Any]: d["prompt"] = self.prompt if self.sink_args: d["sink_args"] = self.sink_args + if self.last_run: + d["last_run"] = self.last_run + if self.next_run: + d["next_run"] = self.next_run + if self.session_ref: + d["session_ref"] = self.session_ref return d @classmethod @@ -70,11 +79,40 @@ def from_toml_dict(cls, name: str, data: Dict[str, Any]) -> "Schedule": sink=data.get("sink", "stdout"), sink_args=data.get("sink_args", {}) or {}, enabled=data.get("enabled", True), + last_run=data.get("last_run"), + next_run=data.get("next_run"), + session_ref=data.get("session_ref"), created_at=data.get("created_at", ""), ) -class ScheduleStore: +@runtime_checkable +class ScheduleStore(Protocol): + """Storage-backend interface for schedules. + + :class:`TomlScheduleStore` is the default, file-backed implementation. The + Protocol exists so an alternative backend (e.g. database-backed) can be + swapped in without changing the daemon or CLI. + """ + + def load(self) -> Dict[str, Schedule]: ... + + def save(self, schedules: Dict[str, Schedule]) -> None: ... + + def add(self, schedule: Schedule) -> None: ... + + def remove(self, name: str) -> None: ... + + def get(self, name: str) -> Schedule: ... + + def set_enabled(self, name: str, enabled: bool) -> Schedule: ... + + def mark_run( + self, name: str, last_run: str, next_run: Optional[str] = None + ) -> Schedule: ... + + +class TomlScheduleStore: """Load/save a collection of :class:`Schedule` objects to a TOML file.""" def __init__(self, path: Path = DEFAULT_STORE_PATH): @@ -93,9 +131,7 @@ def load(self) -> Dict[str, Schedule]: def save(self, schedules: Dict[str, Schedule]) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) - doc = { - "schedules": {s.name: s.to_toml_dict() for s in schedules.values()} - } + doc = {"schedules": {s.name: s.to_toml_dict() for s in schedules.values()}} with open(self.path, "wb") as f: tomli_w.dump(doc, f) @@ -138,3 +174,18 @@ def set_enabled(self, name: str, enabled: bool) -> Schedule: schedules[name].enabled = enabled self.save(schedules) return schedules[name] + + def mark_run( + self, name: str, last_run: str, next_run: Optional[str] = None + ) -> Schedule: + schedules = self.load() + if name not in schedules: + raise KeyError( + f"no schedule named {name!r} in {self.path}; " + f"run `gaia schedule list` to see registered schedules" + ) + schedules[name].last_run = last_run + if next_run is not None: + schedules[name].next_run = next_run + self.save(schedules) + return schedules[name] diff --git a/tests/unit/test_schedule_daemon.py b/tests/unit/test_schedule_daemon.py new file mode 100644 index 000000000..904106b64 --- /dev/null +++ b/tests/unit/test_schedule_daemon.py @@ -0,0 +1,239 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT +"""Unit tests for the schedule daemon (``gaia.schedule.daemon``) and the +store-wiring of :func:`gaia.schedule.runner.fire`. + +Covers: + - ``next_fire_time`` returns an ISO string for a valid cron, None-safe shape. + - ``build_scheduler`` arms only enabled schedules (one job per enabled). + - ``_job`` marks the run on success and logs-but-does-not-raise on failure + (a failing job must not kill the daemon, and must NOT mark the run). + - ``runner.fire`` runs without a real LLM (AgentSDK mocked) and routes the + agent output through ``sinks.dispatch``. + +Hermetic: no network, no real LLM, no filesystem outside ``tmp_path``. +``AgentSDK``/``AgentConfig`` are imported lazily inside ``runner.fire`` so the +patch targets their source module ``gaia.chat.sdk``. +""" + +from __future__ import annotations + +from datetime import datetime + +import pytest + +from gaia.schedule import daemon, runner +from gaia.schedule.store import Schedule, TomlScheduleStore + +# runner.fire imports AgentSDK/AgentConfig lazily from gaia.chat.sdk. +_AGENT_SDK = "gaia.chat.sdk.AgentSDK" +_AGENT_CONFIG = "gaia.chat.sdk.AgentConfig" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_schedule(name: str = "daily", **overrides) -> Schedule: + kwargs = {"name": name, "cron": "0 9 * * *", "prompt": "say hello"} + kwargs.update(overrides) + return Schedule(**kwargs) + + +def _store_with(tmp_path, *schedules) -> TomlScheduleStore: + store = TomlScheduleStore(tmp_path / "schedules.toml") + for s in schedules: + store.add(s) + return store + + +# =========================================================================== +# 1. next_fire_time +# =========================================================================== + + +class TestNextFireTime: + + def test_valid_cron_returns_iso_string(self): + result = daemon.next_fire_time("0 9 * * *") + assert isinstance(result, str) + # Parseable as ISO8601 — the contract the store persists. + datetime.fromisoformat(result) + + def test_every_minute_cron_returns_iso_string(self): + result = daemon.next_fire_time("* * * * *") + assert isinstance(result, str) + datetime.fromisoformat(result) + + +# =========================================================================== +# 2. build_scheduler +# =========================================================================== + + +class TestBuildScheduler: + + def test_arms_only_enabled_schedules(self, tmp_path): + store = _store_with( + tmp_path, + _make_schedule("on", enabled=True), + _make_schedule("off", enabled=False), + ) + scheduler = daemon.build_scheduler(store) + jobs = scheduler.get_jobs() + assert len(jobs) == 1 + assert jobs[0].id == "on" + + def test_empty_store_arms_no_jobs(self, tmp_path): + store = _store_with(tmp_path) + scheduler = daemon.build_scheduler(store) + assert scheduler.get_jobs() == [] + + def test_all_enabled_arms_each(self, tmp_path): + store = _store_with( + tmp_path, + _make_schedule("a", enabled=True), + _make_schedule("b", enabled=True), + ) + scheduler = daemon.build_scheduler(store) + assert {job.id for job in scheduler.get_jobs()} == {"a", "b"} + + +# =========================================================================== +# 3. _job — success and failure paths +# =========================================================================== + + +class TestJob: + + def test_success_marks_run(self, mocker, tmp_path): + store = _store_with(tmp_path, _make_schedule("a")) + sched = store.get("a") + + mock_fire = mocker.patch.object(runner, "fire", return_value="output") + mock_mark = mocker.patch.object(store, "mark_run") + + daemon._job(sched, store) + + mock_fire.assert_called_once_with(sched) + mock_mark.assert_called_once() + # last_run (positional[1]) is an ISO timestamp; next_run is keyword. + call = mock_mark.call_args + assert call.args[0] == "a" + datetime.fromisoformat(call.args[1]) + assert "next_run" in call.kwargs + + def test_success_persists_last_run_via_real_store(self, mocker, tmp_path): + # End-to-end through the real store (only fire is mocked). + store = _store_with(tmp_path, _make_schedule("a")) + sched = store.get("a") + mocker.patch.object(runner, "fire", return_value="output") + + daemon._job(sched, store) + + reloaded = store.get("a") + assert reloaded.last_run is not None + assert reloaded.next_run is not None + + def test_failure_does_not_raise(self, mocker, tmp_path): + store = _store_with(tmp_path, _make_schedule("a")) + sched = store.get("a") + mocker.patch.object(runner, "fire", side_effect=RuntimeError("kaboom")) + mock_mark = mocker.patch.object(store, "mark_run") + + # Must NOT propagate — the daemon stays alive. + daemon._job(sched, store) + + # And must NOT mark the run on failure. + mock_mark.assert_not_called() + + def test_failure_logs_exception(self, mocker, tmp_path): + store = _store_with(tmp_path, _make_schedule("a")) + sched = store.get("a") + mocker.patch.object(runner, "fire", side_effect=RuntimeError("kaboom")) + mock_log = mocker.patch.object(daemon.log, "exception") + + daemon._job(sched, store) + + # Loud failure: the traceback is logged (no silent swallow). + mock_log.assert_called_once() + + def test_failure_leaves_store_unmodified(self, mocker, tmp_path): + store = _store_with(tmp_path, _make_schedule("a")) + sched = store.get("a") + mocker.patch.object(runner, "fire", side_effect=RuntimeError("kaboom")) + + daemon._job(sched, store) + + reloaded = store.get("a") + assert reloaded.last_run is None + assert reloaded.next_run is None + + +# =========================================================================== +# 4. runner.fire — no real LLM, output routed to sink +# =========================================================================== + + +class TestRunnerFire: + + def test_fire_routes_agent_output_to_sink(self, mocker): + mock_sdk_cls = mocker.patch(_AGENT_SDK) + mocker.patch(_AGENT_CONFIG) + mock_sdk_cls.return_value.send.return_value.text = "agent says hi" + mock_dispatch = mocker.patch.object(runner.sinks, "dispatch") + + sched = _make_schedule("a", prompt="do the thing", sink="stdout", sink_args={}) + result = runner.fire(sched) + + assert result == "agent says hi" + # Agent was driven with the schedule's prompt, fresh session. + mock_sdk_cls.return_value.send.assert_called_once_with( + "do the thing", no_history=True + ) + # Output was delivered through the configured sink. + mock_dispatch.assert_called_once_with("stdout", {}, "agent says hi") + + def test_fire_passes_sink_args_to_dispatch(self, mocker): + mock_sdk_cls = mocker.patch(_AGENT_SDK) + mocker.patch(_AGENT_CONFIG) + mock_sdk_cls.return_value.send.return_value.text = "out" + mock_dispatch = mocker.patch.object(runner.sinks, "dispatch") + + sched = _make_schedule( + "a", prompt="p", sink="file", sink_args={"path": "/tmp/x.md"} + ) + runner.fire(sched) + + mock_dispatch.assert_called_once_with("file", {"path": "/tmp/x.md"}, "out") + + def test_fire_skill_only_raises_not_implemented(self, mocker): + # --skill resolution is blocked on #888; fire must fail loudly, never + # reach the agent or the sink. + mock_sdk_cls = mocker.patch(_AGENT_SDK) + mock_dispatch = mocker.patch.object(runner.sinks, "dispatch") + + sched = Schedule(name="s", cron="* * * * *", skill="my-skill") + with pytest.raises(NotImplementedError, match="#888"): + runner.fire(sched) + + mock_sdk_cls.return_value.send.assert_not_called() + mock_dispatch.assert_not_called() + + +# =========================================================================== +# 5. resolve_input +# =========================================================================== + + +class TestResolveInput: + + def test_prompt_returns_prompt_text(self): + sched = _make_schedule("a", prompt="hello prompt") + assert runner.resolve_input(sched) == "hello prompt" + + def test_skill_raises_not_implemented(self): + sched = Schedule(name="s", cron="* * * * *", skill="sk") + with pytest.raises(NotImplementedError, match="skill-format"): + runner.resolve_input(sched) diff --git a/tests/unit/test_schedule_sinks.py b/tests/unit/test_schedule_sinks.py new file mode 100644 index 000000000..4d12fb4eb --- /dev/null +++ b/tests/unit/test_schedule_sinks.py @@ -0,0 +1,214 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT +"""Unit tests for schedule output sinks (``gaia.schedule.sinks``). + +Sinks decide *where* a scheduled run's output goes. Per the GAIA no-fallback +rule, a sink that cannot deliver raises an actionable error — it never silently +swallows the failure. These tests cover the dispatch routing table and each +sink's success + failure paths. + +``requests``, ``platform``, and ``subprocess`` are imported at module scope in +``gaia.schedule.sinks``, so patches target that module (not their origin). +""" + +from __future__ import annotations + +import pytest + +from gaia.schedule import sinks + +# Patch targets — module-scoped imports in gaia.schedule.sinks. +_REQUESTS = "gaia.schedule.sinks.requests" +_PLATFORM_SYSTEM = "gaia.schedule.sinks.platform.system" +_SUBPROCESS_RUN = "gaia.schedule.sinks.subprocess.run" + + +# =========================================================================== +# 1. dispatch routing table +# =========================================================================== + + +class TestDispatchRouting: + + def test_stdout_prints_output(self, capsys): + sinks.dispatch("stdout", {}, "hello world") + captured = capsys.readouterr() + assert captured.out == "hello world\n" + + def test_file_prefix_writes_to_path(self, tmp_path): + target = tmp_path / "log.md" + sinks.dispatch(f"file:{target}", {}, "line one") + assert target.read_text(encoding="utf-8") == "line one\n" + + def test_file_sink_uses_sink_args_path(self, tmp_path): + target = tmp_path / "viaargs.md" + sinks.dispatch("file", {"path": str(target)}, "via args") + assert target.read_text(encoding="utf-8") == "via args\n" + + def test_file_sink_appends(self, tmp_path): + target = tmp_path / "log.md" + sinks.dispatch(f"file:{target}", {}, "first") + sinks.dispatch(f"file:{target}", {}, "second") + assert target.read_text(encoding="utf-8") == "first\nsecond\n" + + def test_file_sink_missing_path_raises(self): + with pytest.raises(ValueError, match="file sink requires a path"): + sinks.dispatch("file", {}, "no path") + + def test_unknown_sink_raises(self): + with pytest.raises(ValueError, match="unknown sink"): + sinks.dispatch("carrier-pigeon", {}, "nope") + + +# =========================================================================== +# 2. file sink — directory creation + write semantics +# =========================================================================== + + +class TestFileSink: + + def test_creates_parent_directories(self, tmp_path): + target = tmp_path / "nested" / "deep" / "log.md" + sinks.dispatch(f"file:{target}", {}, "made the dirs") + assert target.read_text(encoding="utf-8") == "made the dirs\n" + + def test_trailing_newlines_normalized(self, tmp_path): + target = tmp_path / "log.md" + sinks.dispatch(f"file:{target}", {}, "trailing\n\n") + assert target.read_text(encoding="utf-8") == "trailing\n" + + +# =========================================================================== +# 3. telegram sink +# =========================================================================== + + +class TestTelegramSink: + + def test_missing_token_raises(self, monkeypatch): + monkeypatch.delenv("GAIA_TELEGRAM_TOKEN", raising=False) + with pytest.raises(ValueError, match="needs a bot token"): + sinks.dispatch("telegram", {"to": "123"}, "msg") + + def test_missing_recipient_raises(self): + with pytest.raises(ValueError, match="needs a recipient"): + sinks.dispatch("telegram", {"token": "tok"}, "msg") + + def test_token_from_env_is_accepted(self, mocker, monkeypatch): + monkeypatch.setenv("GAIA_TELEGRAM_TOKEN", "env-token") + mock_requests = mocker.patch(_REQUESTS) + mock_requests.post.return_value.status_code = 200 + # No token in sink_args — must fall back to the env var. + sinks.dispatch("telegram", {"to": "123"}, "msg") + mock_requests.post.assert_called_once() + + def test_success_posts_to_send_message(self, mocker): + mock_requests = mocker.patch(_REQUESTS) + mock_requests.post.return_value.status_code = 200 + + sinks.dispatch("telegram", {"token": "tok", "to": "42"}, "hello") + + mock_requests.post.assert_called_once() + url = mock_requests.post.call_args.args[0] + kwargs = mock_requests.post.call_args.kwargs + assert url == f"{sinks.TELEGRAM_API}/bottok/sendMessage" + assert kwargs["json"] == {"chat_id": "42", "text": "hello"} + + def test_non_200_raises_runtime_error(self, mocker): + mock_requests = mocker.patch(_REQUESTS) + resp = mock_requests.post.return_value + resp.status_code = 403 + resp.text = "Forbidden: bot was blocked" + + with pytest.raises(RuntimeError, match="HTTP 403"): + sinks.dispatch("telegram", {"token": "tok", "to": "42"}, "hello") + + def test_request_exception_raises_runtime_error(self, mocker): + import requests as real_requests + + mock_requests = mocker.patch(_REQUESTS) + # Preserve the real exception class so the `except requests.RequestException` + # in production code still matches the raised instance. + mock_requests.RequestException = real_requests.RequestException + mock_requests.post.side_effect = real_requests.RequestException("boom") + + with pytest.raises(RuntimeError, match="could not reach"): + sinks.dispatch("telegram", {"token": "tok", "to": "42"}, "hello") + + +# =========================================================================== +# 4. notification sink +# =========================================================================== + + +class TestNotificationSink: + + def test_unsupported_os_raises_not_implemented(self, mocker): + mocker.patch(_PLATFORM_SYSTEM, return_value="Plan9") + with pytest.raises(NotImplementedError, match="not implemented for 'Plan9'"): + sinks.dispatch("notification", {}, "hi") + + def test_darwin_invokes_osascript(self, mocker): + mocker.patch(_PLATFORM_SYSTEM, return_value="Darwin") + mock_run = mocker.patch(_SUBPROCESS_RUN) + + sinks.dispatch("notification", {"title": "T"}, "body text") + + mock_run.assert_called_once() + args = mock_run.call_args.args[0] + assert args[0] == "osascript" + assert args[1] == "-e" + script = args[2] + assert "display notification" in script + assert "body text" in script + assert "with title" in script + assert "T" in script + assert mock_run.call_args.kwargs["check"] is True + + def test_linux_invokes_notify_send(self, mocker): + mocker.patch(_PLATFORM_SYSTEM, return_value="Linux") + mock_run = mocker.patch(_SUBPROCESS_RUN) + + sinks.dispatch("notification", {"title": "T"}, "body") + + mock_run.assert_called_once() + args = mock_run.call_args.args[0] + assert args == ["notify-send", "T", "body"] + + def test_called_process_error_raises_runtime_error(self, mocker): + import subprocess + + mocker.patch(_PLATFORM_SYSTEM, return_value="Darwin") + mocker.patch( + _SUBPROCESS_RUN, + side_effect=subprocess.CalledProcessError(1, "osascript"), + ) + with pytest.raises(RuntimeError, match="failed to post"): + sinks.dispatch("notification", {}, "hi") + + def test_missing_tool_raises_file_not_found(self, mocker): + mocker.patch(_PLATFORM_SYSTEM, return_value="Linux") + mocker.patch(_SUBPROCESS_RUN, side_effect=FileNotFoundError("no notify-send")) + with pytest.raises(FileNotFoundError, match="notify-send"): + sinks.dispatch("notification", {}, "hi") + + +# =========================================================================== +# 5. _osa AppleScript quoting +# =========================================================================== + + +class TestOsaQuoting: + + def test_wraps_in_double_quotes(self): + assert sinks._osa("plain") == '"plain"' + + def test_escapes_embedded_double_quotes(self): + assert sinks._osa('say "hi"') == '"say \\"hi\\""' + + def test_escapes_backslashes_before_quotes(self): + # Backslash must be escaped first so it does not double-escape the quote. + assert sinks._osa("a\\b") == '"a\\\\b"' + + def test_backslash_and_quote_together(self): + assert sinks._osa('a\\"b') == '"a\\\\\\"b"' diff --git a/tests/unit/test_schedule_store.py b/tests/unit/test_schedule_store.py new file mode 100644 index 000000000..9525f3859 --- /dev/null +++ b/tests/unit/test_schedule_store.py @@ -0,0 +1,312 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT +"""Unit tests for the schedule store (``gaia.schedule.store``). + +Covers the :class:`Schedule` dataclass (TOML round-trip including the new +``last_run`` / ``next_run`` / ``session_ref`` fields, the exactly-one-of +skill/prompt invariant, and the ``created_at`` auto-stamp) and the +:class:`TomlScheduleStore` CRUD surface (add/get/remove/set_enabled/mark_run), +plus the structural :class:`ScheduleStore` Protocol. + +All tests use ``tmp_path`` — no real ``~/.gaia/schedules.toml`` is touched. +""" + +from __future__ import annotations + +import pytest + +from gaia.schedule import TomlScheduleStore as ReexportedTomlStore +from gaia.schedule.store import Schedule, ScheduleStore, TomlScheduleStore + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_schedule(name: str = "daily", **overrides) -> Schedule: + """A valid prompt-based Schedule with sensible defaults for overriding.""" + kwargs = { + "name": name, + "cron": "0 9 * * *", + "prompt": "say hello", + } + kwargs.update(overrides) + return Schedule(**kwargs) + + +def _store(tmp_path) -> TomlScheduleStore: + return TomlScheduleStore(tmp_path / "schedules.toml") + + +# =========================================================================== +# 1. Schedule dataclass — invariants +# =========================================================================== + + +class TestScheduleInvariants: + + def test_prompt_only_is_valid(self): + sched = Schedule(name="p", cron="* * * * *", prompt="hi") + assert sched.prompt == "hi" + assert sched.skill is None + + def test_skill_only_is_valid(self): + sched = Schedule(name="s", cron="* * * * *", skill="my-skill") + assert sched.skill == "my-skill" + assert sched.prompt is None + + def test_both_skill_and_prompt_raises(self): + with pytest.raises(ValueError, match="exactly one of"): + Schedule(name="x", cron="* * * * *", skill="s", prompt="p") + + def test_neither_skill_nor_prompt_raises(self): + with pytest.raises(ValueError, match="exactly one of"): + Schedule(name="x", cron="* * * * *") + + def test_created_at_auto_stamped_when_missing(self): + sched = _make_schedule() + assert sched.created_at # non-empty ISO timestamp + # Parseable as an ISO8601 datetime. + from datetime import datetime + + datetime.fromisoformat(sched.created_at) + + def test_created_at_preserved_when_provided(self): + sched = _make_schedule(created_at="2020-01-01T00:00:00+00:00") + assert sched.created_at == "2020-01-01T00:00:00+00:00" + + def test_new_optional_fields_default_to_none(self): + sched = _make_schedule() + assert sched.last_run is None + assert sched.next_run is None + assert sched.session_ref is None + + +# =========================================================================== +# 2. Schedule TOML round-trip +# =========================================================================== + + +class TestScheduleToToml: + + def test_required_fields_present(self): + d = _make_schedule().to_toml_dict() + assert d["cron"] == "0 9 * * *" + assert d["sink"] == "stdout" + assert d["enabled"] is True + assert d["created_at"] + + def test_prompt_present_skill_absent(self): + d = _make_schedule(prompt="hi").to_toml_dict() + assert d["prompt"] == "hi" + assert "skill" not in d + + def test_skill_present_prompt_absent(self): + d = Schedule(name="s", cron="* * * * *", skill="sk").to_toml_dict() + assert d["skill"] == "sk" + assert "prompt" not in d + + def test_sink_args_omitted_when_empty(self): + d = _make_schedule().to_toml_dict() + assert "sink_args" not in d + + def test_sink_args_present_when_set(self): + d = _make_schedule(sink="file", sink_args={"path": "/tmp/x.md"}).to_toml_dict() + assert d["sink_args"] == {"path": "/tmp/x.md"} + + def test_new_fields_omitted_when_none(self): + d = _make_schedule().to_toml_dict() + assert "last_run" not in d + assert "next_run" not in d + assert "session_ref" not in d + + def test_last_run_present_when_set(self): + d = _make_schedule(last_run="2026-01-01T00:00:00+00:00").to_toml_dict() + assert d["last_run"] == "2026-01-01T00:00:00+00:00" + + def test_next_run_present_when_set(self): + d = _make_schedule(next_run="2026-01-02T00:00:00+00:00").to_toml_dict() + assert d["next_run"] == "2026-01-02T00:00:00+00:00" + + def test_session_ref_present_when_set(self): + d = _make_schedule(session_ref="abc-123").to_toml_dict() + assert d["session_ref"] == "abc-123" + + +class TestScheduleFromToml: + + def test_minimal_dict_defaults(self): + sched = Schedule.from_toml_dict("n", {"cron": "* * * * *", "prompt": "hi"}) + assert sched.name == "n" + assert sched.sink == "stdout" + assert sched.enabled is True + assert sched.sink_args == {} + assert sched.last_run is None + assert sched.next_run is None + assert sched.session_ref is None + + def test_full_round_trip_preserves_all_fields(self): + original = _make_schedule( + name="full", + sink="file", + sink_args={"path": "/tmp/log.md"}, + enabled=False, + last_run="2026-01-01T00:00:00+00:00", + next_run="2026-01-02T00:00:00+00:00", + session_ref="sess-9", + created_at="2025-12-31T00:00:00+00:00", + ) + revived = Schedule.from_toml_dict(original.name, original.to_toml_dict()) + assert revived == original + + def test_from_toml_round_trips_new_fields(self): + data = { + "cron": "* * * * *", + "prompt": "hi", + "last_run": "2026-01-01T00:00:00+00:00", + "next_run": "2026-01-02T00:00:00+00:00", + "session_ref": "sess-1", + } + sched = Schedule.from_toml_dict("n", data) + assert sched.last_run == "2026-01-01T00:00:00+00:00" + assert sched.next_run == "2026-01-02T00:00:00+00:00" + assert sched.session_ref == "sess-1" + + def test_null_sink_args_coerced_to_dict(self): + sched = Schedule.from_toml_dict( + "n", {"cron": "* * * * *", "prompt": "hi", "sink_args": None} + ) + assert sched.sink_args == {} + + +# =========================================================================== +# 3. TomlScheduleStore — persistence + CRUD +# =========================================================================== + + +class TestTomlScheduleStore: + + def test_load_missing_file_returns_empty(self, tmp_path): + store = _store(tmp_path) + assert store.load() == {} + + def test_add_then_get(self, tmp_path): + store = _store(tmp_path) + sched = _make_schedule("a") + store.add(sched) + got = store.get("a") + assert got.name == "a" + assert got.prompt == "say hello" + + def test_add_persists_across_store_instances(self, tmp_path): + path = tmp_path / "schedules.toml" + TomlScheduleStore(path).add(_make_schedule("a")) + # A brand new store instance reads the same file from disk. + reloaded = TomlScheduleStore(path).load() + assert "a" in reloaded + assert reloaded["a"].prompt == "say hello" + + def test_add_duplicate_raises_value_error(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a")) + with pytest.raises(ValueError, match="already exists"): + store.add(_make_schedule("a")) + + def test_get_missing_raises_key_error(self, tmp_path): + store = _store(tmp_path) + with pytest.raises(KeyError, match="gaia schedule list"): + store.get("nope") + + def test_remove(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a")) + store.remove("a") + assert store.load() == {} + + def test_remove_missing_raises_key_error(self, tmp_path): + store = _store(tmp_path) + with pytest.raises(KeyError, match="gaia schedule list"): + store.remove("nope") + + def test_set_enabled_toggles_and_persists(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a", enabled=True)) + returned = store.set_enabled("a", False) + assert returned.enabled is False + # Persisted to disk, not just mutated in memory. + assert store.get("a").enabled is False + + def test_set_enabled_missing_raises_key_error(self, tmp_path): + store = _store(tmp_path) + with pytest.raises(KeyError, match="gaia schedule list"): + store.set_enabled("nope", False) + + def test_save_load_multiple_round_trip(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a")) + store.add(_make_schedule("b", skill="sk", prompt=None)) + loaded = store.load() + assert set(loaded) == {"a", "b"} + assert loaded["b"].skill == "sk" + + +# =========================================================================== +# 4. TomlScheduleStore.mark_run +# =========================================================================== + + +class TestMarkRun: + + def test_mark_run_sets_last_run(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a")) + returned = store.mark_run("a", "2026-06-21T09:00:00+00:00") + assert returned.last_run == "2026-06-21T09:00:00+00:00" + # Persisted. + assert store.get("a").last_run == "2026-06-21T09:00:00+00:00" + + def test_mark_run_sets_next_run_when_provided(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a")) + store.mark_run( + "a", + "2026-06-21T09:00:00+00:00", + next_run="2026-06-22T09:00:00+00:00", + ) + got = store.get("a") + assert got.last_run == "2026-06-21T09:00:00+00:00" + assert got.next_run == "2026-06-22T09:00:00+00:00" + + def test_mark_run_leaves_next_run_untouched_when_omitted(self, tmp_path): + store = _store(tmp_path) + store.add(_make_schedule("a", next_run="2026-01-01T00:00:00+00:00")) + store.mark_run("a", "2026-06-21T09:00:00+00:00") + got = store.get("a") + assert got.last_run == "2026-06-21T09:00:00+00:00" + # next_run was not passed, so the prior value survives. + assert got.next_run == "2026-01-01T00:00:00+00:00" + + def test_mark_run_missing_raises_key_error(self, tmp_path): + store = _store(tmp_path) + with pytest.raises(KeyError, match="gaia schedule list"): + store.mark_run("nope", "2026-06-21T09:00:00+00:00") + + +# =========================================================================== +# 5. ScheduleStore Protocol (structural typing) +# =========================================================================== + + +class TestScheduleStoreProtocol: + + def test_concrete_store_satisfies_protocol(self, tmp_path): + store = _store(tmp_path) + # @runtime_checkable Protocol — structural, no inheritance. + assert isinstance(store, ScheduleStore) + + def test_protocol_not_inherited_by_concrete(self): + # TomlScheduleStore must NOT inherit the Protocol — pure duck typing. + assert ScheduleStore not in TomlScheduleStore.__mro__ + + def test_reexport_is_concrete_store(self): + assert ReexportedTomlStore is TomlScheduleStore