> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Scheduler Module

> Deployment scheduling with provider-agnostic design

# Scheduler Module

The Scheduler module provides deployment scheduling capabilities with a provider-agnostic design.

<Warning>
  Since v4.6.122, the execution primitives (`ScheduledAgentExecutor`, `ShellConditionGate`, `JobResult`) live in `praisonai-bot`. Import them from `praisonai_bot.scheduler.*` in new code. Safety primitives (`RunPolicy`, `PromptScanResult`) stay in the wrapper at `praisonai.scheduler.*`. The old executor/gate paths remain importable as backward-compatibility shims.
</Warning>

## Import

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# Wrapper-tier (praisonai): deployment scheduler + safety primitives
from praisonai.scheduler import ScheduleParser, DeploymentScheduler
from praisonai.scheduler import RunPolicy, PromptScanResult

# Bot-tier (praisonai-bot): execution primitives — canonical since v4.6.122
from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult
from praisonai_bot.scheduler.condition_gate import ShellConditionGate
```

<Tip>
  `from praisonai.scheduler.executor import ScheduledAgentExecutor` and `from praisonai.scheduler.condition_gate import ShellConditionGate` still work as backward-compatibility shims when the `praisonai` wrapper is installed.
</Tip>

## Quick Example

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import DeploymentScheduler

# Create scheduler
scheduler = DeploymentScheduler()

# Schedule deployment every hour
scheduler.schedule(interval_minutes=60)

# Start the scheduler
scheduler.start()
```

## Classes

### `DeploymentScheduler`

Minimal deployment scheduler with provider-agnostic design.

**Features:**

* Simple interval-based scheduling
* Thread-safe operation
* Extensible deployer factory pattern

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import DeploymentScheduler

scheduler = DeploymentScheduler()
scheduler.schedule(interval_minutes=30)
scheduler.start()

# Later, stop the scheduler
scheduler.stop()
```

### `ScheduleParser`

Parses schedule expressions into scheduling parameters. `ScheduleParser` is also re-exported from `praisonai.scheduler.shared` and used internally by `AgentScheduler` / `AsyncAgentScheduler`.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import ScheduleParser

# Parse cron-like expressions
schedule = ScheduleParser.parse("every 30 minutes")
schedule = ScheduleParser.parse("daily at 09:00")
```

### `DeployerInterface`

Abstract interface for deployers to ensure provider compatibility.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import DeployerInterface

class MyDeployer(DeployerInterface):
    def deploy(self) -> bool:
        """Execute deployment. Returns True on success."""
        # Your deployment logic
        return True
```

## Methods

### `DeploymentScheduler.schedule(interval_minutes)`

Schedule deployments at a fixed interval.

**Parameters:**

* `interval_minutes` (int): Minutes between deployments

### `DeploymentScheduler.start()`

Start the scheduler in a background thread.

### `DeploymentScheduler.stop()`

Stop the scheduler.

### `DeploymentScheduler.is_running()`

Check if the scheduler is currently running.

**Returns:** `bool`

## Example: Custom Deployer

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import DeploymentScheduler, DeployerInterface

class DockerDeployer(DeployerInterface):
    def __init__(self, image_name: str):
        self.image_name = image_name
    
    def deploy(self) -> bool:
        import subprocess
        try:
            subprocess.run(
                ["docker", "push", self.image_name],
                check=True
            )
            return True
        except subprocess.CalledProcessError:
            return False

# Use custom deployer
scheduler = DeploymentScheduler(deployer=DockerDeployer("myapp:latest"))
scheduler.schedule(interval_minutes=60)
scheduler.start()
```

## Timezones

A naive `at:` timestamp (no `Z`, no `+HH:MM` offset) and every `cron:` expression resolve their timezone in this order:

| Precedence | Source                                | Where it is set                                                                                                        |
| ---------- | ------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| 1          | `sched.tz` on the schedule            | The `tz` field on `Schedule` / the `tz` argument to `parse_schedule`                                                   |
| 2          | Instance default (`default_timezone`) | `scheduler.timezone` (or `scheduler.tz`) key in `~/.praisonai/config.yaml`, read into the store and passed to `is_due` |
| 3          | `PRAISONAI_SCHEDULE_TIMEZONE`         | Process-wide environment variable                                                                                      |
| 4          | `UTC`                                 | Fallback when none of the above is set                                                                                 |

`resolve_schedule_timezone` in `praisonaiagents.scheduler.due` implements the env-var and UTC fallback; `is_due(job, now, default_timezone=...)` threads the instance default through.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler import parse_schedule
from praisonaiagents.scheduler.due import resolve_schedule_timezone, is_due

# Per-schedule tz — the offset the parser stamps on the Schedule
schedule = parse_schedule("at:2026-03-01T09:00:00", tz="America/New_York")

# Env-var / UTC fallback used when no tz is supplied
tzinfo = resolve_schedule_timezone()  # PRAISONAI_SCHEDULE_TIMEZONE, else UTC
```

```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# ~/.praisonai/config.yaml — process-wide instance default
scheduler:
  timezone: Europe/London
```

<Note>
  An `at:` string that already carries an offset (`at:2026-03-01T09:00:00+05:30`, `at:2026-03-01T09:00:00Z`) is unaffected — the offset in the string always wins. See [Schedule Tools → Timezones](/docs/tools/schedule-tools#timezones-for-at-and-cron).
</Note>

## ScheduleJob — Pre-Run Gate Fields

Since PR #2238, `ScheduleJob` (in `praisonaiagents.scheduler.models`) has two optional fields for the pre-run condition gate:

| Field       | Type            | Default | Description                                                                                          |
| ----------- | --------------- | ------- | ---------------------------------------------------------------------------------------------------- |
| `pre_run`   | `Optional[str]` | `None`  | Shell command run before each tick. Exit 0 + stdout → run (stdout seeds the prompt); non-zero → skip |
| `condition` | `Optional[str]` | `None`  | Advisory natural-language label (round-tripped, not enforced by the default gate)                    |

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler import ScheduleJob, Schedule

job = ScheduleJob(
    name="inbox-watch",
    schedule=Schedule(kind="every", every_seconds=300),
    pre_run="scripts/new_mail.sh",
    condition="new mail",
    message="Summarise these new emails.",
)
```

## Timezones

`parse_schedule` (in `praisonaiagents.scheduler.parser`) resolves the timezone for a naive `at:` timestamp and the clock forms (`at 9am`, `at 17:00`) in order:

1. The `tz=` argument passed to `parse_schedule` / `schedule_add` (per-schedule).
2. The instance default — the `scheduler.timezone` (or `scheduler.tz`) key in `config.yaml`, which `ConfigYamlScheduleStore` reads into its `_default_timezone` and passes to `is_due()`.
3. The `PRAISONAI_SCHEDULE_TIMEZONE` environment variable.
4. The machine's local zone, with the UTC offset in force **on the target date** (DST-correct).

An `at:` string that already carries an offset (`+05:30`, `Z`, `+00:00`) is used verbatim. `cron:` follows the same 1–3 precedence but falls back to **UTC** at step 4.

### Parse-time stamping vs the `is_due()` fallback

As of [PraisonAI #4732](https://github.com/MervinPraison/PraisonAI/pull/4732), the zone is attached **at parse time** by `praisonaiagents.scheduler.due.localize_wall_clock`, so `parse_schedule` returns a `Schedule` whose `at` is already an aware ISO string:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler.parser import parse_schedule

parse_schedule("at:2026-07-01T09:00:00", tz="Europe/London").at  # -> "2026-07-01T09:00:00+01:00"
parse_schedule("at:2026-12-01T09:00:00", tz="Europe/London").at  # -> "2026-12-01T09:00:00+00:00"
```

`is_due()` keeps a legacy branch for naive strings that did not come through the parser — jobs written before this PR, hand-edited `config.yaml` entries, or a `Schedule(kind="at", at=<naive iso>)` built directly. That branch calls the same `localize_wall_clock` helper, so both paths agree. An unknown IANA zone raises `ValueError` from either path instead of firing never.

### The instance default

The store's instance default flows through `is_due(job, now, default_timezone=...)` and applies to naive `at:` values that omit a per-schedule `tz`. Set it in `config.yaml`:

```yaml theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
scheduler:
  timezone: Asia/Kolkata   # naive at: with no per-schedule tz reads/stamps as +05:30
```

`is_due()` accepts the same value directly when you evaluate a job yourself:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler.due import is_due

is_due(job, now, default_timezone="Asia/Kolkata")
```

## ScheduleStoreProtocol — per-principal scoping

Each store method accepts an optional `principal` — the resolved end-user identity used to isolate one gateway user's automations from another's. `None` returns everything (global / single-tenant behaviour), so every call is backward-compatible.

| Method           | Signature                                                    | Behavior when `principal` is given                                                                                                                |
| ---------------- | ------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------- |
| `list`           | `list(agent_id=None, principal=None) -> List[ScheduleJob]`   | Only jobs where `job.principal == principal` are returned. `None` returns everything (pre-scoping global).                                        |
| `get_by_name`    | `get_by_name(name, principal=None) -> Optional[ScheduleJob]` | A job owned by a different identity is treated as **not found** (returns `None`) so a user cannot read another's automation by guessing its name. |
| `remove_by_name` | `remove_by_name(name, principal=None) -> bool`               | A job owned by a different identity is **not** removed (returns `False`) so a user cannot delete another's automation.                            |

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler import get_default_store

store = get_default_store()

alice_jobs = store.list(principal="telegram:12345")     # owner-filtered listing
store.get_by_name("morning-brief", principal="bob")     # -> None if owned by alice
store.remove_by_name("morning-brief", principal="bob")  # -> False; job intact
```

`ScheduleJob.principal` (`Optional[str] = None`) round-trips through `to_dict`/`from_dict` and is omitted when `None`, so existing on-disk store files stay byte-identical.

The same scoping was extended to `SuggestionStore` — `list_pending`, `accept`, and `dismiss` take a `principal` and refuse cross-owner mutations, with the pending cap and dedup window measured per owner. See [Automation Suggestions → Multi-tenant isolation](/docs/docs/features/automation-suggestions#multi-tenant-isolation) for the suggestion-store surface.

## ScheduledAgentExecutor

`ScheduledAgentExecutor` lives in the **bot tier** (`praisonai-bot`). The canonical import is:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai_bot.scheduler import ScheduledAgentExecutor, JobResult
```

<Tip>
  `from praisonai.scheduler.executor import ScheduledAgentExecutor` works as a backward-compatible shim when the `praisonai` wrapper is installed alongside `praisonai-bot`.
</Tip>

### Constructor reference

| Parameter            | Type                             | Default  | Description                                                                                                                         |
| -------------------- | -------------------------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------- |
| `runner`             | `ScheduleRunner`                 | required | An SDK `ScheduleRunner` instance                                                                                                    |
| `agent_resolver`     | `Callable[[Optional[str]], Any]` | required | `(agent_id) -> Agent`; may return `None`                                                                                            |
| `delivery_handler`   | `Optional[Callable[..., Any]]`   | `None`   | Async `(delivery, text) -> None`; routes to a channel bot's `send_message()`                                                        |
| `on_success`         | `Optional[Callable[..., None]]`  | `None`   | Called with `(job, result)` on success                                                                                              |
| `on_failure`         | `Optional[Callable[..., None]]`  | `None`   | Called with `(job, error)` on failure                                                                                               |
| `run_policy`         | `Optional[RunPolicy]`            | `None`   | Wrapper safety gate — scopes tools, scans prompt, persists audit, delivers failure summary                                          |
| `condition_resolver` | `None \| False \| callable`      | `None`   | `None` = auto `ShellConditionGate` when `pre_run` set; `False` = gating disabled; callable `(job) → gate \| None` = custom resolver |

### Public methods

| Method                                             | Return type                | Description                                                           |
| -------------------------------------------------- | -------------------------- | --------------------------------------------------------------------- |
| `async tick()`                                     | `AsyncIterator[JobResult]` | Check for due jobs and execute them, yielding one `JobResult` per job |
| `async tick_all()`                                 | `List[JobResult]`          | Like `tick()` but collects all results into a list                    |
| `async run_loop(interval=15.0, *, max_ticks=None)` | `None`                     | Convenience loop that calls `tick()` at a fixed interval              |

<Note>
  **Atomic claims:** When the backing store supports `claim_due`, each due job is reserved under a cross-process lock so it fires at most once across all tickers/processes/hosts. Stores without that support fall back to the non-atomic `get_due_jobs` path.
</Note>

### Example

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleRunner, FileScheduleStore
from praisonai_bot.scheduler import ScheduledAgentExecutor

agent = Agent(name="assistant", instructions="You are helpful.")
store = FileScheduleStore()
runner = ScheduleRunner(store)

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda agent_id: agent,
)

import asyncio

async def main():
    async for result in executor.tick():
        print(f"Job {result.job.name}: {result.status} ({result.duration:.1f}s)")

asyncio.run(main())
```

### `condition_resolver` values

| Value                           | Behaviour                                                                           |
| ------------------------------- | ----------------------------------------------------------------------------------- |
| `None` (default)                | Auto-uses `ShellConditionGate` for jobs that have `pre_run` set                     |
| `False`                         | Gating disabled for all jobs                                                        |
| `callable(job) -> gate \| None` | Custom resolver — return a `JobConditionProtocol` instance or `None` to skip gating |

## JobConditionProtocol and GateResult

`JobConditionProtocol` and `GateResult` are exported from `praisonaiagents.scheduler`:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents.scheduler import JobConditionProtocol, GateResult
```

* `GateResult` — dataclass with `run: bool`, `context: Optional[str]`, `reason: Optional[str]`
* `JobConditionProtocol` — protocol for custom gate implementations
* `SchedulerProviderProtocol` — trigger backend protocol deciding *when* jobs fire; `ScheduleLoop` (aliased `InProcessScheduleProvider`) is the default. See [Scheduler Providers](/docs/features/scheduler-providers).

### `RunPolicy`

Run-scoped guardrail for unattended scheduled agent runs. Scopes the toolset, scans the assembled prompt for injection patterns, persists a durable output audit, and supports fail-closed delivery on failure.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.scheduler import RunPolicy, PromptScanResult
from praisonai_bot.scheduler import ScheduledAgentExecutor

policy = RunPolicy(
    audit_dir="/var/log/praisonai/runs",
    deliver_on_failure=True,
)

executor = ScheduledAgentExecutor(
    runner=runner,
    agent_resolver=lambda _id: agent,
    run_policy=policy,
)
```

See [Scheduled Run Policy](/docs/features/scheduled-run-policy) for the full reference.

### `PromptScanResult`

Return type from `RunPolicy.scan_prompt()` and custom `scanner` callables.

| Field    | Type            | Description                                |
| -------- | --------------- | ------------------------------------------ |
| `ok`     | `bool`          | `True` if the prompt passed the scan       |
| `reason` | `Optional[str]` | Human-readable reason when `ok` is `False` |

## Related

* [Scheduler Pre-Run Gate](/docs/features/scheduler-pre-run-gate) - Full user-facing documentation
* [Scheduled Run Policy](/docs/features/scheduled-run-policy) - Safety gate for unattended runs
* [praisonai-bot SDK](/docs/sdk/praisonai-bot/index) - Bot-tier package reference
* [Deploy Module](/docs/sdk/praisonai/deploy) - Deployment functionality
* [CLI Scheduler](/docs/cli/scheduler) - CLI scheduling commands