Async Agent Scheduler

Async-native scheduler for running agents with proper cancellation and event loop safety

Async-native agent scheduler that replaces daemon threads with proper async execution and cooperative cancellation.

import asyncio
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

async def main():
    agent = Agent(name="NewsChecker", instructions="Summarise the latest AI news.")
    scheduler = AsyncAgentScheduler(agent=agent, task="Top 3 AI stories")
    await scheduler.start("hourly", run_immediately=True)

asyncio.run(main())

The user schedules recurring agent work; the async scheduler runs jobs with cooperative cancellation.

Quick Start

1. Simple Usage

   Create and start an async scheduler with basic configuration.

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       agent = Agent(
           name="NewsChecker", 
           instructions="Summarise the latest AI news."
       )
   
       scheduler = AsyncAgentScheduler(
           agent=agent, 
           task="Summarise top 3 AI stories"
       )
   
       await scheduler.start("hourly", max_retries=3, run_immediately=True)
   
       # Let it run for 2 hours
       await asyncio.sleep(3600 * 2)
       await scheduler.stop()
   
   asyncio.run(main())

2. With Callbacks and Configuration

   Add success/failure callbacks and custom configuration.

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   def success_callback(result):
       print(f"✅ Success: {result}")
   
   async def async_failure_callback(exception):
       print(f"❌ Async failure: {exception}")
   
   async def main():
       agent = Agent(
           name="DataProcessor",
           instructions="Process and analyze data efficiently."
       )
   
       scheduler = AsyncAgentScheduler(
           agent=agent,
           task="Process latest data batch",
           config={"timeout": 120},
           on_success=success_callback,
           on_failure=async_failure_callback
       )
   
       await scheduler.start("*/30m", max_retries=5)
   
       # Monitor stats
       stats = await scheduler.get_stats_async()
       print(f"Stats: {stats}")
   
       await scheduler.stop()
   
   asyncio.run(main())

3. With Timeout & Budget Limit

   Control execution time and budget with built-in enforcement.

   import asyncio
   from praisonaiagents import Agent
   from praisonai.scheduler import AsyncAgentScheduler
   
   async def main():
       agent = Agent(
           name="CostAwareAgent",
           instructions="Summarise the latest tech headlines."
       )
   
       scheduler = AsyncAgentScheduler(
           agent,
           task="Summarise tech news",
           timeout=30,        # stop a single run after 30s
           max_cost=1.00,     # auto-shutdown once $1.00 spent
       )
   
       await scheduler.start("hourly", run_immediately=True)
       await asyncio.sleep(3600 * 4)
       stats = await scheduler.get_stats_async()
       print(f"Spent ${stats['total_cost_usd']}, remaining ${stats['remaining_budget']}")
       await scheduler.stop()
   
   asyncio.run(main())

How It Works

Daemon state persistence

The scheduler daemon writes catch-up state to ~/.praisonai/schedulers/<name>.json. As of PraisonAI #4537:

• Writes are atomic — the state file is written to a temp file, fsync‘d, then renamed into place with os.replace. A crash mid-write can no longer produce a truncated JSON file that silently disables cron catch-up on next start.
• Corrupt / unreadable state files log a warning — you’ll see Corrupt scheduler state ... or Cannot access scheduler state ... in the logs, then the scheduler re-anchors from process start (previous behaviour with no log).

If a scheduled run misses its slot after a crash, grep the logs for those two messages first. The sync scheduler shares the same behaviour — see Scheduler CLI → Daemon state persistence.

Multi-process safety (at-most-once)

Without an atomic claim, two workers polling the same store could each fire the same due job — causing double-posts, double-charges, or duplicated side effects.

Since PraisonAI PR #2628, FileScheduleStore claims each due job under an OS advisory file lock (fcntl on POSIX, msvcrt on Windows) and hands it back to only one worker per tick. Losers receive an empty list and skip silently.

ConfigYamlScheduleStore — the default store — gained the same atomic claim_due via the fix for PraisonAI#3230. The default gateway/serve configuration is now safe for multi-process deployment without swapping the store.

Under the lock the store re-reads on-disk state, pre-advances last_run_at, and takes a short lease atomically. One-shot jobs stay leased on claim and are removed once their run completes. A claim that never completes is recorded as an abandoned run when its lease expires — a one-shot is retried once, a recurring job runs at its next slot. See Crash Recovery.

from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore

store = FileScheduleStore()
runner = ScheduleRunner(store)

if runner.supports_atomic_claim():
    jobs = runner.claim_due_jobs(owner_id="worker-1", lease_seconds=300.0)
    for job in jobs:
        # ... run the job ...
        runner.complete_run(job.id, owner_id="worker-1")
else:
    jobs = runner.get_due_jobs()

The wrapper’s ScheduledAgentExecutor.tick() does this automatically: it claims under the lock when the store supports it (default owner_id = f"{hostname}:{pid}:{uuid}"), releases the lease after each run, and falls back to get_due_jobs otherwise.

Store write durability

FileScheduleStore never silently loses run history or reports success for a write that didn’t land. Introduced in PraisonAI PR #5901 (fixes #5899).

Cross-process run history

Two stores on the same store_dir — the scheduler daemon and a CLI process, or several gateway workers — can call log_run() concurrently and both records survive.

• log_run() takes the cross-process file lock, reloads history.json, merges newest-first, trims to max_history, and writes atomically.
• If history.json is unreadable (e.g. corrupt JSON), the new record stays in memory and is retried on the next log_run() instead of overwriting saved runs with a partial view.

from praisonaiagents.scheduler import FileScheduleStore

a = FileScheduleStore(store_dir="/tmp/schedules")
b = FileScheduleStore(store_dir="/tmp/schedules")

a.log_run("morning-brief", "succeeded", job_name="brief")
b.log_run("digest", "succeeded", job_name="digest")

# A fresh reader sees both records (newest first)
reader = FileScheduleStore(store_dir="/tmp/schedules")
print([r.job_id for r in reader.get_history()])  # ['digest', 'morning-brief']

Fail-loud mutators

add(job), update(job), remove(job_id), and remove_by_name(name) raise OSError when the on-disk write fails (full disk, EIO, permission denied, serialisation error).

• The store reloads its in-memory jobs from disk first, so list() / get() match what the next restart will see.
• SCHEDULE_ADD / SCHEDULE_REMOVE hooks are not emitted for a failed write — see Schedule Lifecycle Hooks.
• ScheduleRunner.mark_run() is the one exception: it catches and logs OSError from its post-run remove / update. The run already happened, and one bad write must not stop the scheduler loop.

from praisonaiagents.scheduler import FileScheduleStore, ScheduleJob, Schedule

store = FileScheduleStore(store_dir="/mnt/read-only")
job = ScheduleJob(
    name="brief",
    schedule=Schedule(kind="every", every_seconds=3600),
    message="Send Alice her morning brief",
)

try:
    store.add(job)
except OSError as e:
    # SCHEDULE_ADD did not fire, and store.list() does not contain the job
    print(f"schedule not saved: {e}")

Multi-tenant scoping

Attach a principal to a ScheduleJob to bind it to one owner; the store rejects cross-owner reads, name lookups, and deletions.

principal is any stable, opaque string that names the owner — a user id, an email hash, a tenant id. Two owners can hold identically-named jobs without collision or leakage.

from praisonaiagents.scheduler import ScheduleJob, Schedule, FileScheduleStore

store = FileScheduleStore()  # ~/.praisonai/schedules/jobs.json

# Owner "alice" creates a job
store.add(ScheduleJob(
    name="daily-report",
    schedule=Schedule(kind="cron", cron_expr="0 9 * * *"),
    message="Generate the daily report",
    principal="alice",
))

# Owner "bob" makes their own with the same name — no collision, no leak
store.add(ScheduleJob(
    name="daily-report",
    schedule=Schedule(kind="cron", cron_expr="0 17 * * *"),
    message="Generate the end-of-day summary",
    principal="bob",
))

alices_jobs = store.list(principal="alice")                    # only Alice's
alices_report = store.get_by_name("daily-report", principal="alice")

# A guess-by-name from Bob's context removes only Bob's, never Alice's:
store.remove_by_name("daily-report", principal="bob")

Cross-owner access is rejected, not silently merged:

• list(principal="alice") returns only Alice’s jobs.
• get_by_name("daily-report", principal="alice") returns None for a job owned by someone else.
• remove_by_name("daily-report", principal="bob") skips a cross-owner match instead of deleting it.

principal=None on any call preserves the global single-tenant view — used by the CLI and single-user deployments. There is no admin bypass: passing principal="bob" cannot reach Alice’s jobs even when the process runs as root.

Configuration Options

Constructor Parameters

Start Parameters

Stats Response Format

Common Patterns

Pattern 1: Event Loop Safety

The scheduler is safe to construct outside an event loop:

# Safe to create in sync code
scheduler = AsyncAgentScheduler(agent, task)

async def run_later():
    # Async primitives created here
    await scheduler.start("hourly")

Pattern 2: FastAPI Integration

from fastapi import FastAPI
from contextlib import asynccontextmanager

scheduler = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global scheduler
    scheduler = AsyncAgentScheduler(agent, "Background task")
    await scheduler.start("*/10m")
    yield
    await scheduler.stop()

app = FastAPI(lifespan=lifespan)

Pattern 3: Cancellation Handling

async def graceful_shutdown():
    try:
        await scheduler.start("hourly")
    except asyncio.CancelledError:
        print("Scheduler cancelled")
    finally:
        await scheduler.stop()

Pattern 4: Budget-aware Scheduling

import asyncio
from praisonaiagents import Agent
from praisonai.scheduler import AsyncAgentScheduler

async def main():
    agent = Agent(name="ReportAgent", instructions="Generate hourly report")

    # Hard-stop after $5 total spend, 60s per run
    scheduler = AsyncAgentScheduler(
        agent,
        task="Generate report",
        timeout=60,
        max_cost=5.00,
    )

    await scheduler.start("hourly")
    while scheduler.is_running:
        await asyncio.sleep(60)
        stats = await scheduler.get_stats_async()
        if stats["remaining_budget"] is not None and stats["remaining_budget"] < 0.50:
            print(f"⚠️  Budget nearly exhausted: ${stats['remaining_budget']:.4f} left")
    print("Scheduler stopped (budget reached or manual stop)")

asyncio.run(main())

Best Practices

# ✅ Good: Constructor safe in sync code
scheduler = AsyncAgentScheduler(agent, task)

async def later():
    # ✅ Good: Async primitives created here
    await scheduler.start("hourly")

def sync_callback(result):
    print(f"Sync: {result}")

async def async_callback(result):
    await log_to_database(result)

# ✅ Both work seamlessly
scheduler = AsyncAgentScheduler(
    agent=agent,
    task=task,
    on_success=async_callback,
    on_failure=sync_callback
)

# Delay formula: min(max(30, 2 ** attempt), 300) * jitter
# - Floor: ~27s, Cap: 300s
# - Same behavior as sync AgentScheduler
await scheduler.start("hourly", max_retries=5)

try:
    await scheduler.start("hourly")
    await asyncio.sleep(3600)
finally:
    stats = await scheduler.get_stats()
    await scheduler.stop()
    print(f"Final stats: {stats}")

# Production with explicit budget
scheduler = AsyncAgentScheduler(agent, task, max_cost=50.00)

# Disable budget (only when external limits enforce it)
scheduler = AsyncAgentScheduler(agent, task, max_cost=None)

scheduler = AsyncAgentScheduler(
    agent,
    task,
    timeout=30,        # individual run cannot exceed 30s
    max_cost=1.00,
)

# Long-running jobs: hold the claim for the full worst-case run time
runner.claim_due_jobs(owner_id="worker-1", lease_seconds=600.0)

# Fast jobs: shorter lease recovers sooner from a crashed worker
runner.claim_due_jobs(owner_id="worker-1", lease_seconds=60.0)

Skipping Ticks with a Pre-Run Gate

A pre-run gate runs a cheap shell command before each tick and skips the model turn when the check says there’s nothing to do — zero tokens spent on quiet ticks.

See Scheduler Pre-Run Gate for configuration options and examples.

For stateful change-detection — comparing a watched source against the last tick and recording no_change when nothing changed — see Scheduler Change Detection.

Change-detection mode

A monitor spec goes further than a skip: it hashes a watched source each tick and records no_change when nothing changed — “wake only when a source changed, tell me only what changed”.

See Scheduler Monitor for the monitor={"url": ...} / {"command": ...} contract and the succeeded / no_change / failed outcomes.

Related