> ## 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()
```

## 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.",
)
```

## 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

### `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