Scheduler Pre-Run Gate

Skip scheduled model turns when a cheap deterministic check says there's nothing to do

Run a cheap shell check before each scheduled tick — skip the model turn (and any delivery ping) when there’s nothing to do.

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleJob, Schedule

agent = Agent(
    name="InboxWatcher",
    instructions="Summarise new emails and flag anything important.",
)
job = ScheduleJob(
    name="inbox-watch",
    schedule=Schedule(kind="every", every_seconds=300),
    pre_run="scripts/new_mail.sh",
    message="Summarise these new emails and flag anything important.",
)

The user schedules a tick with a pre_run shell hook; the agent only runs when the gate script reports work to do.

Quick Start

1. YAML (simplest)

   Add pre_run to your agents.yaml schedule block. The agent only runs when the script exits 0 with output.

   schedule:
     every: 300
     pre_run: scripts/new_mail.sh
     message: "Summarise these new emails and flag anything important."

2. Python

   Set pre_run directly on a ScheduleJob.

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleJob, Schedule
   
   agent = Agent(
       name="InboxWatcher",
       instructions="Summarise new emails and flag anything important.",
   )
   
   job = ScheduleJob(
       name="inbox-watch",
       schedule=Schedule(kind="every", every_seconds=300),
       pre_run="scripts/new_mail.sh",
       message="Summarise these new emails and flag anything important.",
   )

3. CLI

   Pass --pre-run when adding a scheduled job.

   praisonai schedule add "inbox-watch" \
     -s "*/5m" \
     -m "Summarise new emails" \
     --pre-run "scripts/new_mail.sh" \
     --deliver telegram

How It Works

The gate runs in asyncio.to_thread so a slow check does not block other scheduled ticks. If the gate itself raises an exception, the executor falls back to running the agent (defensive default).

Configuration Options

ScheduleJob fields

ScheduledAgentExecutor parameter

ShellConditionGate options

Gate decision table

Pre-Run Gate vs RunPolicy

Two complementary gates apply on every tick when both are configured:

Common Patterns

Inbox watcher

Only fire when new mail arrives — cuts 288 daily runs to just the count of mail bursts.

schedule:
  every: 300
  pre_run: scripts/new_mail.sh
  message: "Summarise these new emails and flag anything important."

scripts/new_mail.sh exits 0 and prints new messages when mail is present; exits non-zero when the inbox is quiet.

CI babysitter

Alert a channel only when a build breaks.

praisonai schedule add "ci-watch" \
  -s "*/5m" \
  -m "Summarise the CI failure and suggest a fix" \
  --pre-run "scripts/ci_failed.sh" \
  --deliver telegram

Custom Python gate

Use a callable resolver for richer logic than a shell exit code.

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleJob, Schedule, ScheduleRunner, FileScheduleStore
from praisonai_bot.scheduler import ScheduledAgentExecutor

class MyPythonGate:
    def check(self, job):
        import requests
        r = requests.get("https://api.example.com/has-work")
        return r.json().get("pending", 0) > 0

agent = Agent(name="WorkProcessor", instructions="Process pending work items.")

store = FileScheduleStore()
runner = ScheduleRunner(store)

job = ScheduleJob(
    name="work-processor",
    schedule=Schedule(kind="every", every_seconds=60),
    message="Process all pending work items.",
)

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda _: agent,
    condition_resolver=lambda job: MyPythonGate(),
)

Best Practices

# Good — fast local check
scripts/new_mail.sh

# Avoid — slow network calls without caching
scripts/fetch_all_data_and_analyse.sh

from praisonai_bot.scheduler.condition_gate import ShellConditionGate

gate = ShellConditionGate(timeout=5.0)

#!/bin/bash
# scripts/new_mail.sh
# Prints new mail subjects if any, exits non-zero if inbox empty
NEW=$(mail -H 2>/dev/null | head -20)
if [ -z "$NEW" ]; then exit 1; fi
echo "$NEW"

# Good — use environment variable
scripts/check_api.sh

# Avoid — secret visible in job record
curl -H "Authorization: Bearer mysecrettoken" https://api.example.com/check

Skip and Failure Reasons

When a tick is skipped or fails, the JobResult.error field contains one of these reasons:

: <stderr>)” | Gate exit non-zero with stderr | skipped | | “pre-run gate timed out (>{timeout:.0f}s)” | Gate exceeded timeout | skipped | | “pre-run gate error: <exception>” | Gate raised exception (defensive: falls back to run) | — | | “Blocked by run policy: <scan.reason>” | RunPolicy scanner blocked | failed` |

When deliver_on_failure is enabled on the RunPolicy, a failure summary is delivered to the channel:

⚠️ Scheduled job '<name>' failed: <error>

JobResult Reference

Every tick yields a JobResult from praisonai_bot.scheduler:

from praisonai_bot.scheduler import JobResult

Session Routing

session_target on ScheduleJob controls whether a scheduled tick reuses an existing channel session or gets its own isolated context.

session_id is passed only when the resolved agent.chat accepts a session_id keyword argument. The executor inspects the signature before passing it.

schedule:
  every: 300
  session_target: main   # join the channel's ongoing session

Imports: Bot-First and Wrapper Shims

# Canonical (bot-tier)
from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult
from praisonai_bot.scheduler.condition_gate import ShellConditionGate

# Wrapper shims (require pip install praisonai)
from praisonai.scheduler.executor import ScheduledAgentExecutor
from praisonai.scheduler.condition_gate import ShellConditionGate

Related