Scheduled Run Policy

Scope tools, scan prompts, and audit output for unattended scheduled agent runs

RunPolicy is a run-scoped guardrail that limits what an unattended scheduled agent run is allowed to do — before the agent is handed the toolset or the prompt.

from praisonaiagents import Agent
from praisonai_bot.scheduler import ScheduledAgentExecutor  # Tier 2b — bot package
from praisonai.scheduler import RunPolicy                    # Tier 3 — wrapper (safety gate stays here)

agent = Agent(
    name="NewsBot",
    instructions="Summarise today's AI news in 3 bullets.",
)
# executor = ScheduledAgentExecutor(runner=runner, agent_resolver=lambda _: agent, run_policy=RunPolicy())

The user schedules an unattended run; RunPolicy scopes tools, scans the prompt, and audits output before delivery.

Defaults (on by default)

The gateway constructs a default RunPolicy at boot with all protections on — you no longer build one by hand to get the safety net.

The gateway rebuilds the policy from gateway.yaml on every tick, so a live edit plus hot-reload applies on the next tick — no process restart.

Configuring via gateway.yaml

Override the defaults with an optional scheduler: block. Every field is optional; omitted fields keep the fail-closed defaults.

scheduler:
  deliver_on_failure: true          # send a failure summary to the job's delivery target
  scan_prompt: true                 # scan the assembled prompt before the turn
  denied_toolsets:                  # tool categories removed from unattended runs
    - cronjob
    - shell
  audit_dir: /var/log/praisonai/scheduler   # durable full-output audit trail

gateway.yaml or Python RunPolicy?

Gateway deployments tune the policy in gateway.yaml; embedders that run their own executor construct RunPolicy in Python.

Gateway user — tune in gateway.yaml

Add a scheduler: block, hot-reload, and the next tick uses the new policy. A failing job now delivers a summary to its target:

scheduler:
  deliver_on_failure: true
  denied_toolsets:
    - cronjob
    - shell

# hot-reload the running gateway, then trigger a due job once
kill -HUP "$(pgrep -f 'praisonai gateway')"
praisonai schedule tick

A blocked or failing run lands as a failure summary on the job’s delivery target — no silent drop.

Embedder — construct RunPolicy in Python

Import RunPolicy and hand it to your own executor. This is the right path outside the gateway:

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
from praisonai_bot.scheduler import ScheduledAgentExecutor
from praisonai.scheduler.run_policy import RunPolicy, DEFAULT_DENIED_TOOLSETS

agent = Agent(name="NewsBot", instructions="Summarise today's AI news in 3 bullets.")

runner = ScheduleRunner(FileScheduleStore())

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda _id: agent,
    run_policy=RunPolicy(denied_toolsets=DEFAULT_DENIED_TOOLSETS | {"shell"}),
)

Host-app embedding

praisonai.integration.bridges.schedules_runner also builds its executor with a default RunPolicy, so due jobs in an embedded host app are guarded the same way as the gateway.

Start and stop

Call ensure_schedule_runner() once at startup and stop_schedule_runner() on shutdown.

from fastapi import FastAPI
from praisonai.integration.bridges.schedules_runner import (
    ensure_schedule_runner,
    stop_schedule_runner,
)

app = FastAPI()
app.add_event_handler("startup", ensure_schedule_runner)
app.add_event_handler("shutdown", stop_schedule_runner)

Diagnostics

ensure_schedule_runner() reports startup outcomes at three log levels, so operators see when scheduled jobs will not fire.

On a real failure _runner_started remains False, so a subsequent ensure_schedule_runner() call retries automatically.

Enable INFO-level logging to surface the WARNING / EXCEPTION lines:

import logging

logging.getLogger("praisonai.integration.bridges.schedules_runner").setLevel(logging.INFO)
logging.basicConfig(level=logging.INFO)

Quick Start

1. Defaults (safe for unattended runs)

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
   from praisonai_bot.scheduler import ScheduledAgentExecutor
   from praisonai.scheduler import RunPolicy
   
   agent = Agent(
       name="NewsBot",
       instructions="Summarise today's AI news in 3 bullets.",
   )
   
   store = FileScheduleStore()
   runner = ScheduleRunner(store)
   
   executor = ScheduledAgentExecutor(
       runner=runner,
       agent_resolver=lambda _id: agent,
       run_policy=RunPolicy(),
   )

   RunPolicy() with no arguments denies cronjob and messaging-interactive tools and scans every prompt automatically.

2. With audit and fail-closed delivery

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
   from praisonai_bot.scheduler import ScheduledAgentExecutor
   from praisonai.scheduler import RunPolicy
   
   agent = Agent(
       name="ReportBot",
       instructions="Generate a daily summary report.",
   )
   
   store = FileScheduleStore()
   runner = ScheduleRunner(store)
   
   executor = ScheduledAgentExecutor(
       runner=runner,
       agent_resolver=lambda _id: agent,
       run_policy=RunPolicy(
           audit_dir="/var/log/praisonai/runs",
           deliver_on_failure=True,
       ),
   )

   Every run writes full output to /var/log/praisonai/runs. On failure, a compact summary is sent to the delivery target.

3. Strict allow-list with custom scanner

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
   from praisonai_bot.scheduler import ScheduledAgentExecutor
   from praisonai.scheduler import RunPolicy
   
   def my_scanner(prompt):
       return "secret" not in prompt.lower()
   
   agent = Agent(
       name="SecureBot",
       instructions="Process financial data safely.",
   )
   
   store = FileScheduleStore()
   runner = ScheduleRunner(store)
   
   executor = ScheduledAgentExecutor(
       runner=runner,
       agent_resolver=lambda _id: agent,
       run_policy=RunPolicy(
           allowed_toolsets={"search", "summarise"},
           denied_toolsets=set(),
           scanner=my_scanner,
       ),
   )

   Only search and summarise tools are available. The custom scanner replaces the built-in heuristic.

How It Works

What Gets Scanned vs What Doesn’t

Choosing Tool-Scope Mode

Failure Alerting — Once Per Incident

Failure summaries are de-duplicated per incident by default: a broken job alerts once, stays silent on identical repeats, and sends a single recovery note on the next success.

Tolerate a single flaky tick — alert only when the same failure repeats twice.

from praisonaiagents import Agent
from praisonai.scheduler import RunPolicy

agent = Agent(
    name="nightly-brief",
    instructions="Summarise overnight repo activity.",
)

# Tolerate a single flaky tick — alert only when the same failure repeats twice.
# The wrapper folds each failure through the built-in IncidentTracker for you.
policy = RunPolicy(alert_after_failures=2)

agent.start("Every day at 02:00, brief last-night activity to slack:#eng.")

Configuration Options

RunPolicy

PromptScanResult

JobResult Reference

from praisonai_bot.scheduler import JobResult

Built-in Injection Heuristics

The built-in scanner flags these patterns (case-insensitive):

Common Patterns

Deny-list pattern (default behaviour)

from praisonai.scheduler import RunPolicy

policy = RunPolicy()

Blocks cronjob (prevents self-scheduling) and messaging-interactive (no human is present to respond).

Allow-list for sensitive deployments

from praisonai.scheduler import RunPolicy

policy = RunPolicy(
    allowed_toolsets={"search", "summarise"},
)

Only search and summarise are available — everything else is silently removed before the run.

Custom scanner plugin

from praisonai.scheduler import RunPolicy, PromptScanResult

def presidio_scanner(prompt: str) -> PromptScanResult:
    # Delegate to Microsoft Presidio or any guardrail library
    result = presidio_analyzer.analyze(text=prompt, language="en")
    if result:
        return PromptScanResult(ok=False, reason=f"PII detected: {result[0].entity_type}")
    return PromptScanResult(ok=True)

policy = RunPolicy(scanner=presidio_scanner)

Extending the default deny-list

from praisonai.scheduler.run_policy import DEFAULT_DENIED_TOOLSETS, RunPolicy

policy = RunPolicy(
    denied_toolsets=DEFAULT_DENIED_TOOLSETS | {"shell", "filesystem"}
)

Best Practices

RunPolicy(audit_dir="/mnt/persistent/praisonai/runs")

from praisonai.scheduler.run_policy import DEFAULT_DENIED_TOOLSETS

RunPolicy(denied_toolsets=DEFAULT_DENIED_TOOLSETS | {"shell"})

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda _id: agent,
    delivery_handler=my_notify_function,
    run_policy=RunPolicy(deliver_on_failure=True),
)

result = executor.run_job(job)
if result.status == "succeeded" and result.delivery_error:
    print(f"Job ran OK but delivery failed: {result.delivery_error}")
    print(f"Full output saved at: {result.audit_path}")

# As of PraisonAI #4198 the invariant holds both ways:
#   result.delivered is True   ⇔  the payload reached the target
#   result.delivered is False  ⇔  it did not, and delivery_error is populated

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