zeno-scheduler 1.0.0

Zeno proactive-trigger scheduler: one-shot, heartbeat, and cron triggers.

zeno-scheduler

Proactive-trigger scheduler for Zeno.

Fires agent turns autonomously — the assistant doesn't have to wait for the user to send a message. Supports one-shot reminders, periodic heartbeats, and cron-style recurring schedules. Triggers are persisted to SQLite so they survive process restarts. Built-in @tools let the agent itself schedule and cancel its own reminders.

Install

uv add 'zeno-framework[scheduler]'
# or, without the AI package:
uv add zeno-scheduler

Features

• One-shot triggers (schedule_once) — fire at a specific datetime or after a timedelta
• Heartbeat triggers (schedule_heartbeat) — fire every N seconds
• Cron triggers (schedule_cron) — standard 5-field cron expression with timezone support
• Built-in tools so the agent can self-schedule from natural-language reminders: schedule_once, schedule_heartbeat, schedule_cron, list_triggers, cancel_trigger, cancel_trigger_by_name
• (user_id, name) upsert semantics — re-running a seed call with the same name updates the existing row instead of duplicating it
• Missed-fire policies (OnMissed.SKIP / RUN_ONCE / RUN_ALL) for triggers that lapse while the process was down

Minimal usage

import asyncio
from datetime import timedelta
from pathlib import Path

from zeno.agent import Agent
from zeno.app import ZenoApp
from zeno.channels.cli import CliChannel
from zeno.scheduler import Scheduler, SqliteScheduleStore
from zeno.scheduler.tools import schedule_once, list_triggers, cancel_trigger
from zeno.testing import FakeProvider

async def main() -> None:
    store = SqliteScheduleStore(Path("./scheduler.db"))
    scheduler = Scheduler(store)

    # Pre-seed a one-shot reminder. The agent could also call this via
    # the `schedule_once` tool at runtime (which accepts string deltas
    # like "30m" — handy when the model produces them).
    await scheduler.schedule_once(
        user_id="local",
        thread_key=None,
        channel="cli",
        prompt="Reminder: stand up and stretch.",
        delta=timedelta(minutes=30),
        name="stretch-reminder",
    )

    app = ZenoApp(
        agent=Agent(
            name="root",
            instructions="You are a proactive assistant. Schedule reminders when asked.",
            tools=[schedule_once, list_triggers, cancel_trigger],
        ),
        channels=[CliChannel()],
        provider=FakeProvider(),
        scheduler=scheduler,  # ← wires the scheduler into ZenoApp's lifecycle
    )
    await app.run()

asyncio.run(main())

ZenoApp starts and stops the scheduler alongside its other tasks. When a trigger fires, the scheduler calls channel.inbound_put(...) on the matching channel, which delivers the prompt as a synthetic inbound message — the agent processes it through the same turn path as a real user message.

Testing

uv run pytest packages/zeno-scheduler

Tests use an in-memory ScheduleStore and a virtual clock — no real sleep or filesystem I/O.

Part of the Zeno framework. See apps/zeno-example-scheduler for a runnable end-to-end reference wiring.