Metadata-Version: 2.4
Name: zeno-scheduler
Version: 1.0.2
Summary: Zeno proactive-trigger scheduler: one-shot, heartbeat, and cron triggers.
Project-URL: Homepage, https://github.com/nkootstra/zeno
Project-URL: Repository, https://github.com/nkootstra/zeno
Project-URL: Issues, https://github.com/nkootstra/zeno/issues
Project-URL: Changelog, https://github.com/nkootstra/zeno/blob/main/CHANGELOG.md
Author: Niels Kootstra
License-Expression: MIT
License-File: LICENSE
Keywords: agent,ai,cron,scheduler,triggers,zeno
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: aiosqlite<1,>=0.19
Requires-Dist: croniter<7,>=6.0
Requires-Dist: python-ulid>=2.2
Requires-Dist: tzdata>=2024.1
Requires-Dist: zeno-core
Description-Content-Type: text/markdown

# zeno-scheduler

Proactive-trigger scheduler for [Zeno](https://github.com/nkootstra/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 `@tool`s let the
agent itself schedule and cancel its own reminders.

## Install

```bash
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

```python
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

```bash
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](https://github.com/nkootstra/zeno). See
[`apps/zeno-example-scheduler`](https://github.com/nkootstra/zeno/blob/main/apps/zeno-example-scheduler/README.md)
for a runnable end-to-end reference wiring.
