feat(schedule): add gaia schedule CLI + cron-based dispatch

docs/reference/cli.mdx

@@ -1906,6 +1906,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 <name>` | Show one schedule's full configuration and next fire time |
+| `remove <name>` | Delete a schedule |
+| `pause <name>` | Disable a schedule without deleting it |
+| `resume <name>` | Re-enable a paused schedule |
+| `run <name>` | 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:<path>` \| `notification` \| `telegram` |
+| `--to` | string | - | Recipient for the `telegram` sink (Telegram user/chat id) |
+
+**Examples:**
+
+<CodeGroup>
+```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
+```
+</CodeGroup>
+
+### 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:<path>` | 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="<bot-token>"
+gaia schedule add \
+  --name daily-digest \
+  --cron "0 18 * * *" \
+  --prompt "Summarize what changed today" \
+  --sink telegram \
+  --to 123456789
+```
+
+<Note>
+`--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.
+</Note>
+
+---
+
 ## Utility Commands
 
 ### Stats Command

setup.py

@@ -28,6 +28,7 @@
         "gaia.llm.providers",
         "gaia.audio",
         "gaia.chat",
+        "gaia.schedule",
         "gaia.ui",
         "gaia.ui.routers",
         "gaia.database",
@@ -104,6 +105,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'",
         # Required by the `gaia-mcp` bridge (base console_script), which parses
         # multipart uploads via python_multipart at import time. Base — not an
         # extra — so a plain `pip install amd-gaia` ships a working gaia-mcp.

src/gaia/cli.py

@@ -1941,6 +1941,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:<path> | 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")
+
     # Knowledge command — web research via the Tavily wrapper
     knowledge_parser = subparsers.add_parser(
         "knowledge",
@@ -2952,6 +3015,90 @@ def build_parser():
     return parser
 
 
+def _handle_schedule(args):
+    """Dispatch `gaia schedule <action>` (issue #892)."""
+    from gaia.schedule import daemon as schedule_daemon
+    from gaia.schedule import runner as schedule_runner
+    from gaia.schedule.store import Schedule, TomlScheduleStore
+
+    action = getattr(args, "schedule_action", None)
+    store = TomlScheduleStore()
+
+    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":
+        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":
+        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__)
@@ -3124,6 +3271,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 = {

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, TomlScheduleStore
+
+__all__ = ["Schedule", "ScheduleStore", "TomlScheduleStore"]

src/gaia/schedule/daemon.py

@@ -0,0 +1,91 @@
+"""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 datetime import datetime, timezone
+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,
+    TomlScheduleStore,
+)
+
+log = get_logger(__name__)
+
+
+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:
+    """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, store],
+            id=schedule.name,
+            name=schedule.name,
+            replace_existing=True,
+        )
+        armed += 1
+    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 = TomlScheduleStore(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 (store=%s); press Ctrl-C to stop", store.path)
+    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