Scheduler Incidents

Alert once when a scheduled job breaks, not on every tick — with recovery notes and threshold blip-tolerance

Turn a stream of scheduled-run failures into a single de-duped operator alert — one ping when a nightly job breaks, one when it recovers, nothing in between.

Quick Start

1. Just enable it

   Incidents ride on top of the scheduler notepad, so the wrapper attaches the tracker automatically when you run from the scheduler — RunPolicy.alert_after_failures is the one knob you tune from the outside.

   from praisonaiagents import Agent
   
   agent = Agent(
       name="nightly-brief",
       instructions="Summarise overnight repo activity.",
   )
   
   # Deliver runs to Slack; the wrapper attaches incident tracking automatically,
   # so a broken nightly job pings the channel once — not every night at 2 a.m.
   agent.start("Every day at 02:00, brief last-night activity to slack:#eng.")

   from praisonaiagents import Agent
   from praisonai.scheduler import RunPolicy
   
   agent = Agent(
       name="nightly-brief",
       instructions="Summarise overnight repo activity.",
   )
   
   # Blip-tolerance without touching IncidentTracker directly.
   policy = RunPolicy(alert_after_failures=2)

2. Drive ticks yourself

   Use the pure tracker when you drive ticks (custom runner, tests, out-of-process cron).

   from praisonaiagents.scheduler import IncidentTracker
   from praisonaiagents.scheduler.models import RunRecord
   
   tracker = IncidentTracker()          # default: alert on the first failure
   state = {}                            # small dict — persist through JobStateStoreProtocol
   
   # Nightly job fails with the same 429 for three ticks in a row.
   alert = tracker.observe(RunRecord(job_id="nightly", status="failed", error="429 rate limit (req id 0xabc123)"), state)
   if alert:
       deliver_to_operator(f"⚠ Nightly job failing: {alert.error}")
   
   # Next two ticks: identical failure → no re-alert.
   tracker.observe(RunRecord(job_id="nightly", status="failed", error="429 rate limit (req id 0xdef456)"), state)  # None
   tracker.observe(RunRecord(job_id="nightly", status="failed", error="429 rate limit (req id 0x999abc)"), state)  # None
   
   # Next success closes the incident → single recovery note.
   resolved = tracker.observe(RunRecord(job_id="nightly", status="succeeded"), state)
   if resolved:
       deliver_to_operator("✅ Nightly job recovered")

3. Tolerate transient blips

   from praisonaiagents.scheduler import IncidentTracker
   
   # Alert only after the same failure repeats 3 ticks in a row.
   tracker = IncidentTracker(after_failures=3)

How It Works

The scheduler loads per-job state, runs the tick, folds the resulting RunRecord through the tracker, and delivers only when an alert is due.

Curious how this is wired end-to-end? See Scheduled Run Policy → Failure Alerting — Once Per Incident.

What Counts As “The Same” Failure

Failures collapse to one incident when their error_signature matches, and normalise_error strips only the volatile parts of a message before hashing.

The normaliser then keeps a prefix of the first 200 chars and sha256s it.

from praisonaiagents.scheduler import error_signature

# Same signature — only the request id changed.
assert error_signature("429 rate limit (req id 0xabc123def456)") == \
       error_signature("429 rate limit (req id 0x99900012345a)")

# Different signature — a status-code change must re-alert.
assert error_signature("HTTP 404 not found") != \
       error_signature("HTTP 500 server error")

Tolerating Transient Blips

Set after_failures=N to alert only when the same failure repeats N ticks in a row — a single flaky tick that recovers never pages you.

from praisonaiagents.scheduler import IncidentTracker
from praisonaiagents.scheduler.models import RunRecord

tracker = IncidentTracker(after_failures=3)
state = {}

tracker.observe(RunRecord(job_id="poll", status="failed", error="boom"), state)  # None
tracker.observe(RunRecord(job_id="poll", status="failed", error="boom"), state)  # None
alert = tracker.observe(RunRecord(job_id="poll", status="failed", error="boom"), state)  # alerts now
assert alert.count == 3

Pick a threshold with this decision tree:

Wiring the Alert

The returned Incident carries the raw error for the alert body; hand it to the job’s existing delivery target — there is no new delivery API.

from praisonaiagents.scheduler import IncidentTracker
from praisonaiagents.scheduler.models import RunRecord

tracker = IncidentTracker()
state = {}

def deliver_to_operator(text):
    ...  # reuse the job's DeliveryTarget (e.g. slack:#eng)

record = RunRecord(job_id="nightly", job_name="Nightly", status="failed", error="429 rate limit")
incident = tracker.observe(record, state)
if incident:
    deliver_to_operator(f"⚠ {incident.job_name} failing: {incident.error}")

Recovery Notes

On the next success after an alerted incident, the wrapper delivers a single recovery note through the same channel:

✅ Scheduled job '<name>' recovered.

The recovery note is only sent when deliver_on_failure is on — it is the mirror of the failure alert.

Configuration Options

IncidentTracker takes one option.

Each incident is a small dataclass persisted per job.

Common Patterns

Persist tracker state through the store you already use for the notepad — the incident dict round-trips as-is.

from praisonaiagents.scheduler import IncidentTracker
from praisonaiagents.scheduler.models import RunRecord

tracker = IncidentTracker()

def tick(store, record):
    state = store.get_state(record.job_id)   # JobStateStoreProtocol
    incident = tracker.observe(record, state)
    store.set_state(record.job_id, state)    # incident lives inside per-job state
    return incident

Reuse the job’s delivery target for the alert — no second channel.

incident = tracker.observe(record, state)
if incident:
    job.delivery.send(f"⚠ {incident.job_name} failing: {incident.error}")

Raise the threshold for a chatty external API so occasional 429s do not page you.

tracker = IncidentTracker(after_failures=3)

Best Practices

Related