zeph-scheduler

Cron-based periodic and one-shot task scheduler with SQLite persistence for Zeph.

Overview

Manages recurring and deferred background tasks. Periodic tasks run on a cron schedule; one-shot tasks fire at a specific point in time. All job state, next-run timestamps, and task mode are persisted in SQLite. The scheduler is controlled at runtime via an mpsc channel — tasks can be added or cancelled without restarting the agent. When combined with the experiments feature flag, the scheduler can run autonomous experiment sessions on a cron schedule via TaskKind::Experiment. Task prompts are injected with an explicit execution prefix for unambiguous agent-loop recognition. Feature-gated behind scheduler.

Key Modules

• scheduler — Scheduler event loop; evaluates due tasks on each tick, drains the SchedulerMessage channel, and dispatches execution to registered handlers
• store — JobStore for SQLite-backed job persistence (upsert, record_run, mark_done, delete, next_run management)
• task — ScheduledTask, TaskDescriptor, TaskHandler, TaskKind, TaskMode — core type definitions
• handlers — CustomTaskHandler — injects a sanitized prompt into the agent loop via mpsc::Sender<String>
• sanitize — sanitize_task_prompt — strips control characters and truncates to 512 code points
• update_check — UpdateCheckHandler for GitHub releases version check
• error — SchedulerError error types

Task Modes

TaskMode controls when a task fires:

[!NOTE] One-shot tasks are automatically removed from the in-memory task list and marked done in the store after they execute. Upsert an existing name to update a task in place.

Runtime Control via SchedulerMessage

The Scheduler exposes an mpsc::Sender<SchedulerMessage> returned from Scheduler::new(). The LLM (via SchedulerExecutor) and other subsystems send messages on this channel to add or cancel tasks without touching the scheduler loop directly.

pub enum SchedulerMessage {
    Add(Box<TaskDescriptor>),
    Cancel(String),   // task name
}

Messages are drained at the start of every tick. The channel capacity is 64 slots; try_send is used to avoid blocking.

Built-in Tasks

CustomTaskHandler

CustomTaskHandler implements TaskHandler and forwards the task field from the job config as a sanitized prompt string to the agent loop via mpsc::Sender<String>. It is safe to call when the channel is full or closed — both conditions log a warning and return Ok(()).

use tokio::sync::mpsc;
use zeph_scheduler::{CustomTaskHandler, Scheduler, ScheduledTask, TaskKind};

let (agent_tx, agent_rx) = mpsc::channel(16);
let handler = CustomTaskHandler::new(agent_tx);
scheduler.register_handler(&TaskKind::Custom("my_task".into()), Box::new(handler));

sanitize_task_prompt

User-supplied task prompts pass through sanitize_task_prompt before being injected into the agent loop. The function strips control characters below U+0020 (except \n and \t) and truncates to 512 Unicode code points.

use zeph_scheduler::sanitize_task_prompt;

let safe = sanitize_task_prompt("hello\x00\x01world\nok");
assert_eq!(safe, "helloworld\nok");

UpdateCheckHandler

UpdateCheckHandler implements TaskHandler and queries the GitHub releases API to compare the running version against the latest published release. When a newer version is detected it sends a human-readable notification over an mpsc::Sender<String> channel.

use tokio::sync::mpsc;
use zeph_scheduler::{ScheduledTask, Scheduler, TaskKind, UpdateCheckHandler};

let (tx, rx) = mpsc::channel(4);
let handler = UpdateCheckHandler::new(env!("CARGO_PKG_VERSION"), tx);

let task = ScheduledTask::new(
    "update_check",
    "0 0 9 * * *",   // daily at 09:00
    TaskKind::UpdateCheck,
    serde_json::Value::Null,
)?;
scheduler.add_task(task);
scheduler.register_handler(&TaskKind::UpdateCheck, Box::new(handler));

Notification format sent via the channel:

New version available: v0.13.0 (current: v0.12.0).
Update: https://github.com/bug-ops/zeph/releases/tag/v0.12.0

Behaviour on error (network failure, non-2xx response, oversized body, parse error, invalid semver) — logs a warn message and returns Ok(()).

Configuration

Use Scheduler::with_max_tasks(store, shutdown_rx, max) to set the limit at construction time. Pass tick_interval_secs to run_with_interval():

use tokio::sync::watch;
use zeph_scheduler::{JobStore, Scheduler};

let store = JobStore::open("scheduler.db").await?;
let (_, shutdown_rx) = watch::channel(false);
let (mut scheduler, task_tx) = Scheduler::with_max_tasks(store, shutdown_rx, 200);
scheduler.init().await?;
scheduler.run_with_interval(30).await;  // tick every 30 seconds

PERF-SC-04 Fix

Previously, a periodic task with a missing next_run value in the store would fire immediately on the next tick regardless of its cron schedule. The fix: when next_run is NULL, the scheduler computes and persists the next occurrence from the cron expression and skips the current tick. Tasks now only fire when next_run <= now.

JobStore Schema

CREATE TABLE IF NOT EXISTS scheduled_jobs (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    name      TEXT NOT NULL UNIQUE,
    cron_expr TEXT NOT NULL DEFAULT '',
    kind      TEXT NOT NULL,
    last_run  TEXT,
    next_run  TEXT,
    status    TEXT NOT NULL DEFAULT 'pending',
    task_mode TEXT NOT NULL DEFAULT 'periodic',
    run_at    TEXT
)

task_mode is 'periodic' or 'oneshot'. run_at holds the ISO 8601 UTC timestamp for one-shot tasks. The init() method applies ALTER TABLE migrations for older schemas that lack task_mode and run_at.

Installation

cargo add zeph-scheduler

Enabled via the scheduler feature flag on the root zeph crate.

[!IMPORTANT] Requires Rust 1.88 or later.

Documentation

Full documentation: https://bug-ops.github.io/zeph/

License

MIT