Scheduler Change Detection

Turn a scheduled job into a stateful monitor that only spends tokens when a watched source has changed

A monitor job hashes a watched source each tick and only runs the model when that source has changed — quiet ticks record no_change and spend zero tokens.

from praisonaiagents.scheduler import Schedule, ScheduleJob

job = ScheduleJob(
    name="status-watch",
    schedule=Schedule(kind="every", every_seconds=300),
    monitor={"command": "curl -s https://api.example.com/status"},
    message="Summarise any changes to the status endpoint.",
)

The first tick runs and seeds a baseline; later ticks compare the fresh hash to the last one and stay silent until something actually changes.

Quick Start

1. Watch a source

   Point monitor at a cheap source. Each tick probes it, hashes the output, and only runs the model on a change:

   from praisonaiagents.scheduler import Schedule, ScheduleJob
   
   job = ScheduleJob(
       name="status-watch",
       schedule=Schedule(kind="every", every_seconds=300),
       monitor={"command": "curl -s https://api.example.com/status"},
       message="Summarise any changes to the status endpoint.",
   )

2. Pin the model for unattended runs

   An unattended job created against a cheap default should not silently follow a later, pricier one — snapshot the model and pin it:

   from praisonaiagents.scheduler import Schedule, ScheduleJob
   
   job = ScheduleJob(
       name="status-watch",
       schedule=Schedule(kind="every", every_seconds=300),
       monitor={"url": "https://api.example.com/status"},
       message="Summarise any changes.",
       provider="openai",
       model="gpt-4o-mini",
       pin_model=True,  # default — the run fails closed if the model drifts
   )

Which Gate Should I Use?

Four distinct controls decide whether — and how — a scheduled tick runs.

Run Statuses

A monitor tick records exactly one status so operators can tell “unchanged” apart from “gate said skip”.

no_change is distinct from skipped on purpose: it means “the watched source did not change”, not “the gate said don’t run”.

Model-Free Delivery Action

Set command to run a shell command on schedule and deliver its stdout verbatim — no agent, no LLM call.

from praisonaiagents.scheduler import Schedule, ScheduleJob

job = ScheduleJob(
    name="disk-watchdog",
    schedule=Schedule(kind="every", every_seconds=3600),
    command="df -h",
    command_timeout=60.0,  # kill the command after this many seconds
)

The command’s process group is killed if it exceeds command_timeout, and that tick is recorded as failed — a hung command can never stall the ticker.

Custom Stateful Gates

Implement JobConditionProtocol.should_run to build your own change-detection gate. A stateful gate accepts the job’s prior state and returns a GateResult that both suppresses an unchanged source and carries the next-tick hash.

import hashlib
from praisonaiagents.scheduler import GateResult

class UrlHashGate:
    def should_run(self, job, *, state=None):
        state = state or {}
        fresh = fetch(job.monitor["url"])  # your bounded fetch
        digest = hashlib.sha256(fresh.encode()).hexdigest()
        if state.get("hash") == digest:
            return GateResult(no_change=True, state_updates={"hash": digest})
        return GateResult(run=True, context=fresh, state_updates={"hash": digest})

When no_change=True, GateResult forces run=False in __post_init__, so a gate can never fire a tick the contract defines as suppressed. Leaving state_updates=None writes nothing — an error probe leaves prior state untouched so a later recovery still suppresses.

Out of the Box

The default wrapper store ConfigYamlScheduleStore already implements JobStateStoreProtocol — a job picks up state persistence automatically, with no custom store needed. The executor loads prior state into a “Notepad” prompt block and persists updates back under a 16 KiB per-job cap. See Cross-Run Memory for the full notepad walkthrough.

Custom State Stores

Roll your own store (for tests or a custom backend) by implementing JobStateStoreProtocol to persist a job’s per-tick scratchpad — a last-seen hash, a cursor, or a watermark.

class InMemoryStateStore:
    def __init__(self):
        self._state = {}

    def get_state(self, job_id):
        return self._state.get(job_id, {})

    def set_state(self, job_id, state):
        self._state[job_id] = state

    def clear_state(self, job_id):
        self._state.pop(job_id, None)

All three methods are optional for a store to provide — callers detect support with hasattr() and treat absence as today’s stateless behaviour.

Configuration Options

Fields on ScheduleJob that drive monitor mode and model pinning.

GateResult fields for stateful gates:

Best Practices

Related