Task Scheduling

The tasks module covers Aleph's time-based scheduling: cron for recurring agent jobs, heartbeat for periodic external probes with L1→L2 escalation, and a shared infrastructure layer (clock abstraction, delivery, schedule parsing, persistence, reaper) that both subsystems use. The daemon process owns the timer loops and is the canonical host for cron and heartbeat RPCs.

Design Philosophy

The scheduling system follows three principles:

1. Testable time — Every time-dependent module accepts a Clock trait so tests inject a FakeClock and the production code uses SystemClock (Utc::now()).
2. Atomic persistence — Both CronStore and HeartbeatStore are SQLite-backed with an in-memory cache; changes are applied in memory and persisted on persist(), so the on-disk row always matches the live view the timer loop saw.
3. Push, not poll — Successful mutations publish CronJobChanged / HeartbeatTaskChanged on the gateway event bus so the panel can drop its refresh loop.

Cron Jobs

The tasks::cron module schedules recurring jobs. A CronService wraps an internal ServiceState and exposes the public async API that gateway handlers and CLI tools call.

┌──────────────────────────────────────────────────────────┐
│                     CronService                           │
├──────────────────────────────────────────────────────────┤
│  state: Arc<ServiceState<SystemClock>>                    │
│    ├─ store:  Arc<Mutex<CronStore>>  (SQLite + in-mem)    │
│    ├─ clock:  Arc<dyn Clock>                              │
│    └─ config: CronConfig                                  │
│  event_bus: Option<Arc<GatewayEventBus>>  (CronJobChanged)│
└──────────────────────────────────────────────────────────┘
            │                               ▲
            ▼                               │
   gateway handlers ◄─────handle_*──────────┘
   (cron.list, cron.get, cron.create, cron.update,
    cron.delete, cron.status, cron.run, cron.runs,
    cron.toggle)

Schedule kinds

A CronJob carries one of three schedule kinds (src/tasks/cron/config.rs):

pub enum ScheduleKind {
    /// Fire at a specific timestamp.
    At { at: i64, delete_after_run: bool },
    /// Fire every N milliseconds (with optional anchor).
    Every { every_ms: i64, anchor_ms: Option<i64> },
    /// Standard 6-field cron expression with optional tz / stagger.
    Cron { expr: String, tz: Option<String>, stagger_ms: Option<i64> },
}

Public API

impl CronService {
    pub fn new(config: CronConfig) -> Result<Self, String>;
    pub fn with_event_bus(self, bus: Arc<GatewayEventBus>) -> Self;

    pub async fn list_jobs(&self) -> Result<Vec<CronJobView>, String>;
    pub async fn get_job(&self, id: &str) -> Result<CronJobView, String>;

    pub async fn add_job(&self, job: CronJob) -> Result<String, String>;
    pub async fn update_job(&self, id: &str, updates: CronJobUpdates) -> Result<(), String>;
    pub async fn delete_job(&self, id: &str) -> Result<(), String>;

    pub async fn enable_job(&self, id: &str) -> Result<(), String>;
    pub async fn disable_job(&self, id: &str) -> Result<(), String>;
    pub async fn toggle_job(&self, id: &str) -> Result<bool, String>;
    pub async fn run_job(&self, id: &str) -> Result<(), String>;

    pub async fn reap_history(&self) -> Result<u64, String>;
}

Features

• Schedule kinds — one-shot (At), interval (Every), and standard 6-field cron expressions with optional timezone and per-job stagger.
• JSON atomic persistence with in-memory cache; persist() writes the in-memory working copy to SQLite on every mutation.
• Concurrent execution with a configurable per-process cap; jobs can also pin a SessionTarget::Main (same session) or Isolated (default) session.
• Job chaining — on_success / on_failure triggers compose jobs into DAGs; cycles are refused at config time.
• Failure alerting — FailureAlertConfig (after N consecutive failures, cooldown, target) plus a global notify_on_failure_default policy that defaults unconfigured jobs to alert the originating channel.
• Run history — every run lands in cron_runs with started_at, completed_at, status (Ok / Error / Skipped / Timeout), ErrorReason::Transient|Permanent, and the triggering source (Schedule / Chain / Manual / Catchup).
• Template rendering — tasks::cron::template substitutes variables into job prompts before agent invocation.
• Webhook delivery — tasks::cron::webhook_target lets a job post its outcome to an arbitrary HTTP endpoint.
• Carry-over sweeper — tasks::cron::carryover runs a process-global sweep (OnceLock-guarded, no double-spawn across hot-reload) to resume interrupted jobs.

Clock abstraction

The cron service depends on the shared Clock trait (src/tasks/shared/clock.rs):

pub trait Clock: Send + Sync + 'static {
    fn now_ms(&self) -> i64;
    fn now_utc(&self) -> DateTime<Utc> { /* default from now_ms */ }
}

SystemClock is the production implementation; FakeClock (in the testing submodule) provides deterministic time control for unit tests.

RPCs (real implementations)

All cron.* RPCs registered in src/bin/aleph-server/commands/start/builder/handlers/agents.rs are backed by real handlers (src/gateway/handlers/cron/real.rs). The earlier *_stub placeholders that lived in src/gateway/handlers/cron/stubs.rs are no longer reached on the hot path:

Heartbeat Probes

The tasks::heartbeat module runs periodic L1 probes with optional L2 agent analysis when a probe fires. The same CronService-style pattern — HeartbeatService facade, HeartbeatServiceState, HeartbeatStore — keeps the two subsystems symmetric.

┌──────────────────────────────────────────────────────────────┐
│                     HeartbeatService                          │
├──────────────────────────────────────────────────────────────┤
│  state: Arc<HeartbeatServiceState>                            │
│    ├─ store: Arc<Mutex<HeartbeatStore>>                       │
│    └─ config: HeartbeatConfig (tick interval, max concurrent, │
│                                 per-job timeout, dedup)       │
│  wake_queue: Arc<WakeQueue>                                   │
│  event_bus: Option<Arc<GatewayEventBus>> (HeartbeatTaskChanged)│
└──────────────────────────────────────────────────────────────┘

Features

• Probe execution — tasks::heartbeat::executor runs each task's configured ProbeConfig (tool + params + trigger condition).
• Wake queue — tasks::heartbeat::wake lets an external caller poke a wake signal into the timer loop without blocking on the per-task mutex.
• Deduplication — tasks::heartbeat::dedup collapses repeated outcomes within a DedupConfig { window_ms, similarity_threshold, max_history } so noisy probes do not flood the model.
• Run history — tasks::heartbeat::history persists past runs for inspection via heartbeat.runs.
• Configurable intervals and timeouts — HeartbeatConfig::tick_interval_secs, max_concurrent, job_timeout_secs.

Public API

impl HeartbeatService {
    pub fn new(store: HeartbeatStore, config: HeartbeatConfig) -> Self;
    pub fn with_event_bus(self, bus: Arc<GatewayEventBus>) -> Self;

    pub async fn list_tasks(&self) -> Vec<HeartbeatTaskView>;
    pub async fn get_task(&self, id: &str) -> Option<HeartbeatTaskView>;

    pub async fn add_task<C: Clock>(&self, task: HeartbeatTask, clock: &C) -> Result<String, String>;
    pub async fn update_task<C: Clock>(&self, id: &str, updates: HeartbeatTaskUpdates, clock: &C)
        -> Result<(), String>;
    pub async fn delete_task(&self, id: &str) -> Result<(), String>;
    pub async fn toggle_task<C: Clock>(&self, id: &str, clock: &C) -> Result<bool, String>;

    pub async fn reap_history(&self) -> Result<usize, String>;
    pub fn request_shutdown(&self);
}

RPCs (real implementations)

All heartbeat.* RPCs registered in agents.rs are real handlers (src/gateway/handlers/heartbeat.rs); the older *_stub definitions remain in the file as regression tests but are no longer dispatched by the registry:

Task Infrastructure

The tasks::shared module provides cross-cutting infrastructure consumed by both cron and heartbeat:

• clock — Clock trait, SystemClock, FakeClock (in testing).
• schedule — Pure schedule-kind parsing and next-run computation.
• delivery — DeliveryEngine / DeliveryPayload / DeliveryTarget (how cron results land in a channel or webhook).
• store — Reusable SQLite helpers (open, schema migration, in-memory cache).
• targets — Delivery target type definitions (channels, webhooks).
• active_hours — Quiet-hours and active-hours gating for jobs.
• retry_hint — Retry policy primitives (transient vs permanent classification).
• reaper — Periodically calls cron.reap_history / heartbeat.reap_history so cron_runs / heartbeat history rows do not grow without bound.

Safety Properties

• No underflow — Time arithmetic uses saturating_sub / saturating_mul (Clock::now_utc uses from_timestamp_millis so pre-epoch values round-trip correctly).
• No SQL injection — All queries use parameterised params![].
• UTF-8 safe — String truncation uses char_indices()-aware helpers.
• No static mut — All shared state goes through crate::sync_primitives.

Code Location

Cron:

• src/tasks/cron/mod.rs — CronService facade
• src/tasks/cron/config.rs — CronConfig, CronJob, ScheduleKind, RunStatus, ErrorReason, SessionTarget, TriggerSource, FailureAlertConfig, JobStateV2, JobRun
• src/tasks/cron/store.rs — CronStore (SQLite + in-memory cache)
• src/tasks/cron/service/ — ServiceState, ops (CRUD), timer (next-due loop), concurrency, catchup
• src/tasks/cron/executor.rs — Job execution
• src/tasks/cron/chain.rs — Job chaining (on_success / on_failure)
• src/tasks/cron/alert.rs — Failure alerting
• src/tasks/cron/template.rs — Prompt template rendering
• src/tasks/cron/webhook_target.rs — Webhook delivery
• src/tasks/cron/stagger.rs — Hash-based cron stagger
• src/tasks/cron/history.rs — cron_runs schema and helpers
• src/tasks/cron/carryover.rs — Carry-over sweeper

Heartbeat:

• src/tasks/heartbeat/mod.rs — HeartbeatService facade
• src/tasks/heartbeat/config.rs — HeartbeatConfig, HeartbeatTask, HeartbeatTaskView, ProbeConfig, TriggerCondition, DedupConfig
• src/tasks/heartbeat/store.rs — HeartbeatStore
• src/tasks/heartbeat/service/ — HeartbeatServiceState, ops, timer
• src/tasks/heartbeat/executor.rs — Probe execution
• src/tasks/heartbeat/probe.rs — Probe dispatcher
• src/tasks/heartbeat/dedup.rs — Outcome deduplication
• src/tasks/heartbeat/history.rs — Run history
• src/tasks/heartbeat/wake.rs — WakeQueue

Shared:

• src/tasks/shared/clock.rs — Clock trait + SystemClock + FakeClock
• src/tasks/shared/schedule.rs — Pure scheduling computation
• src/tasks/shared/delivery.rs — DeliveryEngine and friends
• src/tasks/shared/store.rs — SQLite helpers
• src/tasks/shared/targets.rs — Delivery target types
• src/tasks/shared/active_hours.rs — Active-hours gating
• src/tasks/shared/retry_hint.rs — Retry classification
• src/tasks/shared/reaper.rs — History reaper daemon

RPCs:

• src/gateway/handlers/cron/real.rs — real cron.* handlers
• src/gateway/handlers/cron/stubs.rs — legacy stubs (kept for regression tests, not dispatched)
• src/gateway/handlers/heartbeat.rs — real heartbeat.* handlers (the file also contains the legacy *_stub definitions as regression tests)
• src/bin/aleph-server/commands/start/builder/handlers/agents.rs — registers cron.* and heartbeat.* on the gateway server

See Also

• Daemon — Owns the timer loops, day-boundary log rotation, ALEPH-BOOT marker, sleep inhibitor, and the Guardian Judge proactive-wake decisions that drive cron and heartbeat in practice.