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

# Process Scheduler Tools

> Schedule and automate recurring tasks, workflows, and data operations with flexible cron-based, interval-based, and one-time scheduling capabilities.

## **Overview**

### **Tool Name**

```json theme={null}
process_scheduler_tools
```

### **Purpose**

The **process\_scheduler\_tools** enable scheduling and automation of recurring tasks within Genesis Data Agents. Create scheduled jobs that execute prompts, trigger workflows, run data pipelines, and perform automated operations at specified intervals or times. Perfect for ETL automation, report generation, monitoring tasks, data refreshes, and any operation that needs to run on a schedule.

#### **Functions Available**

1. **`scheduler_action`**: *`Manage scheduled jobs including creation, execution, monitoring, and history tracking with support for cron, interval, and one-time scheduling.`*

### Key Features

<Steps>
  <Step title="Flexible Scheduling" icon="calendar-days">
    Support for cron expressions, interval-based scheduling, and one-time execution dates.
  </Step>

  <Step title="Prompt Execution" icon="message-bot">
    Schedule natural language prompts to be executed by specified bots at scheduled times.
  </Step>

  <Step title="Job Management" icon="tasks">
    Create, list, pause, resume, modify, and delete scheduled jobs with full lifecycle control.
  </Step>

  <Step title="Execution History" icon="clock-rotate-left">
    Track job run history with timestamps, outputs, and execution status for auditing.
  </Step>

  <Step title="Manual Triggers" icon="play">
    Execute scheduled jobs on-demand without waiting for the next scheduled time.
  </Step>

  <Step title="Error Handling" icon="triangle-exclamation">
    Automatic retry logic, failure notifications, and detailed error logging.
  </Step>
</Steps>

For scheduled thread-title attribution in Costs, see
[Cost accounting](/docs/Setup/Advanced-Settings#cost-accounting).

## **Input Parameters for Each Function**

### **`scheduler_action`**

**Parameters**

| **Name** | **Definition** | **Format** |
| - | - | :-: |
| **action** | Scheduler operation to perform. Values: `STATUS`, `ADD_JOB`, `LIST_JOBS`, `GET_JOB`, `REMOVE_JOB`, `PAUSE_JOB`, `RESUME_JOB`, `RUN_JOB`, `MODIFY_JOB`, `GET_HISTORY`, `GET_RUN`, `CLEAR_HISTORY`, `KILL_RUN`. | String (required) |
| **job\_id** | Unique identifier for the scheduled job. Required for most job-specific operations. | String |
| **what\_to\_do\_prompt** | Natural language prompt describing the task to execute. This is the ACTUAL TASK, not a scheduling request. | String |
| **trigger** | Scheduling configuration as JSON object. Types: `cron`, `interval`, `date`. | Object |
| **job\_bot\_id** | Bot ID that will execute the scheduled job (defaults to current bot). | String |
| **job\_thread\_id** | Thread ID for job execution context (optional). | String |
| **name** | Human-readable name for the job (optional). | String |
| **coalesce** | Whether to run missed jobs as single execution (default: False). | Boolean |
| **max\_instances** | Maximum concurrent instances of job (default: 1). | Integer |
| **misfire\_grace\_time** | Seconds to wait before considering job misfired (default: None). | Integer |
| **run\_id** | Specific run identifier for retrieving execution details or killing runs. | String |
| **limit** | Maximum number of history records to return (default: 100). | Integer |
| **offset** | Number of history records to skip for pagination (default: 0). | Integer |

<Tip>
  The **what\_to\_do\_prompt** should describe the actual task to perform, not the scheduling itself. For example: "Check for new data and update the dashboard" NOT "Schedule a check every 5 minutes".
</Tip>

### **Expiration and intentional perpetual schedules**

New and changed recurring schedules must include an `end_date` when they run
more often than hourly. This applies to prompt jobs and Flow schedules.

| Minimum time between runs | Expiration requirement |
| - | - |
| Less than 5 minutes | A future `end_date` no more than 7 days from creation or renewal. |
| At least 5 minutes, less than 1 hour | A future `end_date` no more than 30 days from creation or renewal. |
| At least 1 hour, less than 1 day | `end_date` is optional; review the schedule every 90 days. |
| 1 day or longer | `end_date` remains optional. |

Put `end_date` inside the `trigger` object as an ISO timestamp. Include a UTC
offset, or use the trigger's timezone (`UTC` by default). A future `start_date`
does not extend these limits. For cron schedules, the tier uses the shortest
time-of-day gap, including midnight, even when weekday or calendar restrictions
make actual runs farther apart. The scheduler stops recurring execution after
the end date; an already running job can finish.

For an intentional indefinite poller, use `perpetual: true` and a nonempty
`perpetual_reason` inside `trigger`. Explain who or what owns stopping it.
`perpetual: true` and `end_date` cannot be combined. Managed Jira and ServiceNow
pollers record this intent because their enabled polling configuration owns
their lifetime; disabling polling removes their schedules.

Dataset refresh policies also record perpetual intent for ongoing schedules.
The dataset and dashboard refresh editor prefills an editable lifetime reason
for new schedules. Reserved refresh job IDs do not waive expiration validation;
an existing refresh edited without an end date must explicitly justify its
ongoing operation. Internal periodic backups retain their separate lifecycle.

`GET_JOB` and `LIST_JOBS` expose `end_date`, `perpetual`, and `perpetual_reason`
alongside the complete `trigger_config`, so operators can audit expiry and
indefinite intent. Existing jobs survive startup without a new expiration;
every `MODIFY_JOB` request validates their current or proposed schedule,
including name-only and prompt-only edits. To edit a legacy frequent job, include
a qualifying future `end_date` or explicit perpetual intent in the same request.
The scheduler editor and prompt-job update API apply the same rule. Pausing,
resuming, running immediately, and removing jobs do not require new expiration
intent. Unrelated Flow detail edits also preserve the existing schedule.

Renew an existing job with an end-date-only trigger update. `MODIFY_JOB` merges
these fields with its stored trigger, preserving the cadence and timezone:

```python theme={null}
from datetime import datetime, timedelta, timezone

scheduler_action(
    action="MODIFY_JOB",
    job_id="data_freshness_check",
    trigger={
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)
```

<Tip>
  Choose the slowest cadence that meets the task. Frequent LLM-backed jobs can
  spend tokens and hit rate limits; use a direct Flow or event-driven automation
  when the work does not need a new LLM decision on every run. Expiration requires
  explicit renewal; no automatic renewal or pre-expiry reminder is created.
</Tip>

## **Use Cases**

1. **Automated Data Refreshes**
   Schedule regular data extracts, transformations, and loads to keep dashboards and reports current.

2. **Report Generation**
   Generate and distribute daily, weekly, or monthly reports automatically at specified times.

3. **System Monitoring**
   Check system health, data quality, pipeline status, or SLA compliance at regular intervals.

4. **Data Quality Checks**
   Run validation jobs to detect anomalies, missing data, or schema changes on a schedule.

5. **Batch Processing**
   Schedule resource-intensive operations during off-peak hours to optimize system performance.

## **Workflow/How It Works**

1. **Step 1: Check Scheduler Status**
   Verify scheduler is running and get statistics:
   ```python theme={null}
   scheduler_action(action="STATUS")
   ```

2. **Step 2: Create Cron-Based Job**
   Schedule a job using cron expression (runs every day at 9 AM):
   ```python theme={null}
   scheduler_action(
       action="ADD_JOB",
       job_id="daily_sales_report",
       name="Daily Sales Report Generator",
       what_to_do_prompt="Generate sales report for yesterday and send to the team",
       trigger={
           "type": "cron",
           "hour": 9,
           "minute": 0
       }
   )
   ```

3. **Step 3: Create Interval-Based Job**
   Schedule a job to run every 30 minutes for the next 7 days:
   ```python theme={null}
   from datetime import datetime, timedelta, timezone

   scheduler_action(
       action="ADD_JOB",
       job_id="data_freshness_check",
       name="Data Freshness Monitor",
       what_to_do_prompt="Check if source data has been updated in the last hour and alert if stale",
       trigger={
           "type": "interval",
           "minutes": 30,
           "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
       }
   )
   ```

4. **Step 4: Create One-Time Job**
   Schedule a job for specific date/time:
   ```python theme={null}
   scheduler_action(
       action="ADD_JOB",
       job_id="quarterly_analysis",
       name="Q1 2024 Analysis",
       what_to_do_prompt="Perform comprehensive Q1 analysis and generate executive summary",
       trigger={
           "type": "date",
           "run_date": "2024-04-01T08:00:00"
       }
   )
   ```

5. **Step 5: List All Jobs**
   View all scheduled jobs:
   ```python theme={null}
   scheduler_action(action="LIST_JOBS")
   ```

6. **Step 6: Get Job Details**
   Retrieve specific job configuration:
   ```python theme={null}
   scheduler_action(
       action="GET_JOB",
       job_id="daily_sales_report"
   )
   ```

7. **Step 7: Run Job Manually**
   Trigger immediate execution:

   ```python theme={null}
   scheduler_action(
       action="RUN_JOB",
       job_id="daily_sales_report"
   )
   ```

   An immediate run resumes a paused schedule. If saving that state fails, the
   response reports a partial failure: execution may already have started,
   and restart uses the last saved state. Check job history before retrying.

8. **Step 8: View Execution History**
   Check job run history:
   ```python theme={null}
   scheduler_action(
       action="GET_HISTORY",
       job_id="daily_sales_report",
       limit=10
   )
   ```

9. **Step 9: Pause/Resume Jobs**
   Temporarily disable or re-enable jobs:
   ```python theme={null}
   # Pause job
   scheduler_action(
       action="PAUSE_JOB",
       job_id="data_freshness_check"
   )

   # Resume job
   scheduler_action(
       action="RESUME_JOB",
       job_id="data_freshness_check"
   )
   ```

10. **Step 10: Modify Existing Job**
    Update job configuration:
    ```python theme={null}
    scheduler_action(
        action="MODIFY_JOB",
        job_id="daily_sales_report",
        trigger={
            "type": "cron",
            "hour": 8,
            "minute": 30
        }
    )
    ```

## **Integration Relevance**

* **data\_connector\_tools** to schedule automated data extraction and loading operations.
* **airflow\_tools** to trigger Airflow DAGs on schedule or as part of workflow orchestration.
* **dbt\_action** to schedule dbt model runs for regular data transformations.
* **github\_connector\_tools** / **git\_and\_file\_tools** (`git_action`, for GitLab) to schedule automated commits or deployments.
* **slack\_tools** to send scheduled notifications, reports, or alerts.
* **project\_manager\_tools** to schedule task execution and mission progress updates.

## **Configuration Details**

* **Scheduler Type**: APScheduler with persistent job store (survives restarts).
* **Time Zone**: Jobs run in UTC by default; specify timezone in cron/date triggers.
* **Persistence**: Job definitions and history stored in database.
* **Concurrency**: Configurable max\_instances per job (default: 1).
* **Missed Jobs**: Coalesce option determines if missed jobs run once or skipped.
* **Grace Period**: misfire\_grace\_time allows late job execution within window.
* **Job Storage**: Job run outputs and logs retained based on retention policy.

### **Direct scheduler jobs (Flow, dataset, dashboard)**

Genesis also registers **direct** scheduler jobs that bypass the LLM prompt
path. Job IDs use fixed prefixes:

| Prefix | Example job ID | Action |
| - | - | - |
| `flow_` | `flow_daily_sales` | Run the Flow with `trigger=schedule`. |
| `dataset_refresh_` | `dataset_refresh_my_dataset` | Refresh one dataset snapshot. |
| `dashboard_refresh_` | `dashboard_refresh_my_dashboard` | Refresh one dashboard. |

Flow schedules created through `flow_action(SCHEDULE)` or the Flows UI sync into
`flow_<flow_id>` jobs automatically.

When a direct Flow job fails after the runner finalizes the run, Genesis can post
an in-app **Scheduled Flow Failed** notification (see
[Unattended failure notifications](/docs/User-Guide/Flows#unattended-failure-notifications)).
The scheduler subprocess records `failure_notified_downstream: true` when the
Flow runner reports the operator is covered for this failure streak — either
because a notification was delivered on this run (`operator_notified`) or
because the runner deliberately stayed quiet on a repeat failure in the same
streak (`operator_notification_coverage` is `suppressed`). It then skips its
generic job-failure bell for that run. If the Flow never reached the runner
(deleted Flow, backend unreachable), or coverage is `not_covered`, no
downstream notification is recorded and the generic scheduler alert still fires.

### **Trigger Types**

#### **Cron Trigger**

Schedule jobs using cron-like expressions:

```python theme={null}
{
    "type": "cron",
    "hour": 9,              # Hour (0-23)
    "minute": 0,            # Minute (0-59)
    "day_of_week": "mon",   # Day of week (mon, tue, wed, thu, fri, sat, sun or 0-6)
    "day": 1,               # Day of month (1-31)
    "month": "*",           # Month (1-12)
    "timezone": "UTC"       # Optional timezone
}
```

**Cron Examples:**

* Every day at 9 AM: `{"type": "cron", "hour": 9, "minute": 0}`
* Every Monday at 8:30 AM: `{"type": "cron", "hour": 8, "minute": 30, "day_of_week": "mon"}`
* First day of every month: `{"type": "cron", "hour": 6, "minute": 0, "day": 1}`
* Every hour: `{"type": "cron", "minute": 0}`
* Every 15 minutes: `{"type": "cron", "minute": "*/15", "end_date": "<future ISO timestamp within 30 days>"}`

#### **Interval Trigger**

Schedule jobs at regular intervals:

```python theme={null}
{
    "type": "interval",
    "seconds": 30,      # Optional: interval in seconds
    "minutes": 5,       # Optional: interval in minutes
    "hours": 2,         # Optional: interval in hours
    "days": 1           # Optional: interval in days
}
```

**Interval Examples:**

* Every 30 seconds: `{"type": "interval", "seconds": 30, "end_date": "<future ISO timestamp within 7 days>"}`
* Every 5 minutes: `{"type": "interval", "minutes": 5, "end_date": "<future ISO timestamp within 30 days>"}`
* Every 2 hours: `{"type": "interval", "hours": 2}`
* Every day: `{"type": "interval", "days": 1}`

#### **Date Trigger**

Schedule one-time job execution:

```python theme={null}
{
    "type": "date",
    "run_date": "2024-04-01T08:00:00"  # ISO format date/time
}
```

<Warning>
  For cron triggers, be mindful of timezone settings. All times default to UTC unless explicitly specified. Convert local times to UTC or specify timezone in trigger configuration.
</Warning>

## **Limitations or Notes**

1. **Execution Environment**: Jobs execute in bot context with access to bot's tools and permissions.
2. **Long-Running Jobs**: Jobs with execution time > 5 minutes may be terminated (configure timeout if needed).
3. **Concurrency**: max\_instances controls how many copies of a job can run simultaneously.
4. **Missed Executions**: If scheduler is down during scheduled time, coalesce determines behavior.
5. **History Retention**: Job run history retained for configurable period (default: 90 days).
6. **Job Limits**: Recommended maximum of 100 active jobs per bot for performance.
7. **Prompt Complexity**: Keep what\_to\_do\_prompt clear and focused; complex multi-step operations may need breaking down.
8. **Thread Context**: If job\_thread\_id not specified, each run creates new thread context.

### **Supported Actions**

✅ **STATUS** - Get scheduler status and statistics\
✅ **ADD\_JOB** - Create new scheduled job\
✅ **LIST\_JOBS** - List all scheduled jobs\
✅ **GET\_JOB** - Get specific job details\
✅ **REMOVE\_JOB** - Delete scheduled job\
✅ **PAUSE\_JOB** - Temporarily disable job\
✅ **RESUME\_JOB** - Re-enable paused job\
✅ **RUN\_JOB** - Execute job immediately\
✅ **MODIFY\_JOB** - Update job configuration\
✅ **GET\_HISTORY** - View job execution history\
✅ **GET\_RUN** - Get specific run details and output\
✅ **CLEAR\_HISTORY** - Delete old history records\
✅ **KILL\_RUN** - Terminate running job execution

### **Not Supported**

❌ Sub-second scheduling (minimum interval: 1 second)\
❌ Conditional triggers based on external events (use separate monitoring job)\
❌ Cross-bot job dependencies (implement in job prompt logic)\
❌ Job priority or queue management\
❌ Distributed job execution across multiple scheduler instances\
❌ Job chaining or workflow DAGs (use Airflow for complex workflows)\
❌ Dynamic schedule modification during execution

### **Output**

* **ADD\_JOB**: Job ID, next run time, and creation confirmation.
* **LIST\_JOBS**: Table of jobs with ID, name, trigger type, next run time, and status.
* **GET\_JOB**: Complete job configuration including trigger, bot assignment, and metadata.
* **GET\_HISTORY**: List of execution records with run ID, start time, end time, status, and output summary.
* **GET\_RUN**: Complete run details including full output, error messages, and execution context. For Flow runs interrupted by a server restart, verify the target before relying on or rerunning a step; see [Flow run status and interrupted steps](/docs/User-Guide/Flows#a-run-started-from-a-chat-reports-back-to-it).
* **RUN\_JOB**: Run ID and confirmation of manual execution trigger.
* **STATUS**: Scheduler health, active jobs count, running jobs, and pending executions.
* **Errors**: Detailed error messages with troubleshooting guidance.

## **Best Practices**

<CardGroup cols={2}>
  <Card title="Clear Prompts" icon="message">
    Write specific, actionable prompts. Good: "Query sales table and email summary". Bad: "Do sales stuff".
  </Card>

  <Card title="Appropriate Intervals" icon="clock">
    Don't over-schedule. Balance freshness needs with system load. Consider impact of frequent executions.
  </Card>

  <Card title="Error Handling" icon="shield-exclamation">
    Include error handling instructions in prompts: "If query fails, log error and notify team".
  </Card>

  <Card title="Monitor History" icon="chart-line">
    Regularly review execution history to catch failures, performance issues, or unexpected behavior.
  </Card>

  <Card title="Use Descriptive Names" icon="tag">
    Name jobs clearly to indicate purpose. Makes management and troubleshooting easier.
  </Card>

  <Card title="Test Before Scheduling" icon="flask">
    Run jobs manually first to verify they work correctly before setting them on a schedule.
  </Card>
</CardGroup>

## **Example: Complete Scheduling Workflow**

```python theme={null}
# Step 1: Check scheduler is running
status = scheduler_action(action="STATUS")
print(f"Scheduler status: {status}")

# Step 2: Create daily data refresh job (runs at 2 AM UTC)
scheduler_action(
    action="ADD_JOB",
    job_id="daily_customer_refresh",
    name="Daily Customer Data Refresh",
    what_to_do_prompt="""
Execute the following data refresh workflow:
1. Query the customer database for updates since last run
2. Transform data according to business rules
3. Load into analytics warehouse
4. Run data quality checks
5. If any issues found, send alert to #data-ops Slack channel
6. On success, update last_run timestamp
""",
    trigger={
        "type": "cron",
        "hour": 2,
        "minute": 0
    },
    coalesce=True,
    max_instances=1
)

# Step 3: Create frequent monitoring job (every 5 minutes)
from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="pipeline_health_check",
    name="Pipeline Health Monitor",
    what_to_do_prompt="""
Check health of all active data pipelines:
1. Query pipeline_status table for jobs running longer than expected
2. Check for any jobs in 'failed' state in the last hour
3. Verify data freshness for critical tables (< 2 hours old)
4. If any issues detected, send alert with details
5. Log check results to monitoring table
""",
    trigger={
        "type": "interval",
        "minutes": 5,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    },
    max_instances=1
)

# Step 4: Create weekly report job (every Monday at 9 AM)
scheduler_action(
    action="ADD_JOB",
    job_id="weekly_executive_report",
    name="Weekly Executive Report",
    what_to_do_prompt="""
Generate weekly executive report:
1. Calculate KPIs for previous week (revenue, customers, growth metrics)
2. Generate trend charts for key metrics
3. Create summary document with insights
4. Save report to shared drive
5. Send email notification with report link to executive team
6. Post summary to #exec-updates Slack channel
""",
    trigger={
        "type": "cron",
        "hour": 9,
        "minute": 0,
        "day_of_week": "mon"
    }
)

# Step 5: Create end-of-month job (runs on the 1st at midnight)
scheduler_action(
    action="ADD_JOB",
    job_id="monthly_close_process",
    name="Monthly Financial Close",
    what_to_do_prompt="""
Execute month-end financial close process:
1. Run month-end reconciliation queries
2. Generate financial statements
3. Calculate month-over-month variance analysis
4. Archive previous month data
5. Initialize new month tracking tables
6. Send completion notification to finance team
""",
    trigger={
        "type": "cron",
        "hour": 0,
        "minute": 0,
        "day": 1
    },
    max_instances=1
)

# Step 6: List all scheduled jobs
jobs = scheduler_action(action="LIST_JOBS")
print(f"Total scheduled jobs: {len(jobs['jobs'])}")
for job in jobs['jobs']:
    print(f"- {job['name']}: next run at {job['next_run_time']}")

# Step 7: Run a job manually for testing
print("\\nTesting pipeline health check...")
run_result = scheduler_action(
    action="RUN_JOB",
    job_id="pipeline_health_check"
)
print(f"Manual run triggered: {run_result['run_id']}")

# Wait for execution to complete
import time
time.sleep(5)

# Step 8: Check execution results
run_details = scheduler_action(
    action="GET_RUN",
    run_id=run_result['run_id']
)
print(f"Run status: {run_details['status']}")
print(f"Output: {run_details['output']}")

# Step 9: View execution history for a job
history = scheduler_action(
    action="GET_HISTORY",
    job_id="pipeline_health_check",
    limit=5
)
print(f"\\nLast 5 executions of pipeline health check:")
for run in history['runs']:
    print(f"- {run['start_time']}: {run['status']} ({run['duration']}s)")

# Step 10: Modify job schedule (change to every 10 minutes instead of 5)
scheduler_action(
    action="MODIFY_JOB",
    job_id="pipeline_health_check",
    trigger={
        "type": "interval",
        "minutes": 10
    }
)
print("Updated health check to run every 10 minutes")

# Step 11: Pause job temporarily (e.g., during maintenance)
scheduler_action(
    action="PAUSE_JOB",
    job_id="daily_customer_refresh"
)
print("Paused daily refresh job for maintenance")

# Later: Resume the job
scheduler_action(
    action="RESUME_JOB",
    job_id="daily_customer_refresh"
)
print("Resumed daily refresh job")

# Step 12: Clean up old history (keep last 30 days)
scheduler_action(
    action="CLEAR_HISTORY",
    job_id="pipeline_health_check",
    older_than_days=30
)
print("Cleared history older than 30 days")
```

## **Advanced Features**

### **Conditional Execution**

Implement conditional logic in job prompts:

```python theme={null}
from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="smart_refresh",
    name="Conditional Data Refresh",
    what_to_do_prompt="""
Execute smart data refresh:
1. Check if source data has been updated since last run
2. If no new data, skip processing and log skip reason
3. If new data exists:
   a. Calculate approximate processing time based on data volume
   b. If > 10k new records, send notification that large batch is processing
   c. Execute refresh
   d. Validate results
   e. Send completion summary
""",
    trigger={
        "type": "interval",
        "minutes": 15,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)
```

### **Multi-Step Workflows**

Chain multiple operations in a single scheduled job:

```python theme={null}
scheduler_action(
    action="ADD_JOB",
    job_id="etl_pipeline",
    name="Complete ETL Pipeline",
    what_to_do_prompt="""
Execute complete ETL pipeline:

EXTRACT:
1. Pull data from production database (last 24 hours)
2. Extract data from API endpoints
3. Download files from S3 bucket

TRANSFORM:
4. Validate data quality (check for nulls, duplicates, schema)
5. Apply business logic transformations
6. Aggregate metrics by date and region
7. Enrich with reference data

LOAD:
8. Load to staging tables
9. Run merge to production tables
10. Update materialized views
11. Refresh BI tool cache

VALIDATE & NOTIFY:
12. Run post-load validation queries
13. Compare record counts before/after
14. Generate execution summary
15. Send success notification with metrics
16. If any step fails, rollback and alert immediately
""",
    trigger={
        "type": "cron",
        "hour": 3,
        "minute": 0
    },
    max_instances=1
)
```

### **Dynamic Scheduling**

Create jobs programmatically based on configuration:

```python theme={null}
# Define multiple similar jobs from configuration
data_sources = [
    {"name": "customers", "table": "dim_customers", "hour": 1},
    {"name": "orders", "table": "fact_orders", "hour": 2},
    {"name": "products", "table": "dim_products", "hour": 3},
    {"name": "inventory", "table": "fact_inventory", "hour": 4}
]

for source in data_sources:
    scheduler_action(
        action="ADD_JOB",
        job_id=f"refresh_{source['name']}",
        name=f"{source['name'].title()} Data Refresh",
        what_to_do_prompt=f"""
Refresh {source['name']} data:
1. Query source system for {source['name']} updates
2. Stage in temp table
3. Merge to {source['table']}
4. Log row counts and execution time
5. Update metadata table with refresh timestamp
""",
        trigger={
            "type": "cron",
            "hour": source['hour'],
            "minute": 0
        }
    )
    print(f"Created refresh job for {source['name']}")
```

### **Error Recovery Jobs**

Create monitoring jobs that check for and recover from failures:

```python theme={null}
from datetime import datetime, timedelta, timezone

scheduler_action(
    action="ADD_JOB",
    job_id="error_recovery",
    name="Automatic Error Recovery",
    what_to_do_prompt="""
Check for and recover from pipeline errors:
1. Query error_log table for failures in last hour
2. For each failed job:
   a. Check if error is transient (connection timeout, lock, etc.)
   b. If transient, attempt retry with exponential backoff
   c. If persistent, escalate to on-call engineer
   d. Log recovery attempt and outcome
3. Update error_log with recovery status
4. If multiple failures of same type, create incident ticket
""",
    trigger={
        "type": "interval",
        "minutes": 15,
        "end_date": (datetime.now(timezone.utc) + timedelta(days=7)).isoformat()
    }
)
```

### **Scheduled Cleanup Jobs**

Maintain system health with automated cleanup:

```python theme={null}
scheduler_action(
    action="ADD_JOB",
    job_id="nightly_cleanup",
    name="Nightly System Cleanup",
    what_to_do_prompt="""
Perform system maintenance and cleanup:
1. Archive logs older than 90 days
2. Delete temporary tables created > 7 days ago
3. Vacuum and analyze frequently updated tables
4. Clear cached query results older than 24 hours
5. Remove orphaned files from staging areas
6. Compact transaction log files
7. Generate cleanup summary report
8. Log space reclaimed and performance improvements
""",
    trigger={
        "type": "cron",
        "hour": 1,
        "minute": 30
    }
)
```

## **Monitoring & Alerting**

### **Track Job Health**

```python theme={null}
def monitor_scheduled_jobs():
    """Monitor health of all scheduled jobs"""
    # Get all jobs
    jobs = scheduler_action(action="LIST_JOBS")
    
    issues = []
    
    for job in jobs['jobs']:
        # Get recent history
        history = scheduler_action(
            action="GET_HISTORY",
            job_id=job['job_id'],
            limit=5
        )
        
        # Check for consecutive failures
        recent_runs = history['runs']
        if len(recent_runs) >= 3:
            if all(run['status'] == 'failed' for run in recent_runs[:3]):
                issues.append({
                    'job_id': job['job_id'],
                    'issue': 'Three consecutive failures',
                    'severity': 'high'
                })
        
        # Check for long-running jobs
        for run in recent_runs:
            if run['status'] == 'running' and run['duration'] > 3600:
                issues.append({
                    'job_id': job['job_id'],
                    'issue': f'Job running for {run["duration"]}s',
                    'severity': 'medium'
                })
        
        # Check for stale jobs (not run in expected timeframe)
        if job['last_run_time']:
            from datetime import datetime
            last_run = datetime.fromisoformat(job['last_run_time'])
            now = datetime.now()
            hours_since = (now - last_run).total_seconds() / 3600
            
            # For hourly jobs, alert if no run in 3 hours
            if 'interval' in job['trigger'] and hours_since > 3:
                issues.append({
                    'job_id': job['job_id'],
                    'issue': f'No execution in {hours_since:.1f} hours',
                    'severity': 'medium'
                })
    
    return issues

# Run monitoring
issues = monitor_scheduled_jobs()
if issues:
    print(f"Found {len(issues)} issues:")
    for issue in issues:
        print(f"⚠️ {issue['job_id']}: {issue['issue']} [{issue['severity']}]")
else:
    print("✅ All scheduled jobs healthy")
```

## **Troubleshooting**

<AccordionGroup>
  <Accordion title="Job Not Executing at Expected Time">
    * Verify job is not paused (check status with GET\_JOB)
    * Check scheduler status (use STATUS action)
    * Verify trigger configuration is correct
    * Check for timezone mismatches (cron times are UTC by default)
    * Review misfire\_grace\_time setting
    * Check if max\_instances limit is reached
  </Accordion>

  <Accordion title="Job Execution Fails Immediately">
    * Review error message in GET\_RUN output
    * Verify what\_to\_do\_prompt is valid and clear
    * Check if bot has required permissions and tools
    * Test prompt manually before scheduling
    * Review job execution history for patterns
    * Ensure bot\_id specified is valid and active
  </Accordion>

  <Accordion title="Job Runs But Produces Wrong Results">
    * Review full output using GET\_RUN action
    * Check if prompt is ambiguous or unclear
    * Verify input data and assumptions are correct
    * Test prompt manually to reproduce issue
    * Add more specific instructions to prompt
    * Review execution context and variables
  </Accordion>

  <Accordion title="Missed Job Executions">
    * Check if scheduler was running during scheduled time
    * Review coalesce setting (determines missed job behavior)
    * Check system logs for scheduler downtime
    * Verify misfire\_grace\_time is appropriate
    * Consider using interval triggers for critical jobs
    * Monitor scheduler health proactively
  </Accordion>

  <Accordion title="Cannot Add New Job">
    * Verify job\_id is unique (not already in use)
    * Check trigger configuration format is valid
    * Ensure what\_to\_do\_prompt is provided
    * Verify bot\_id exists if specified
    * Check for scheduler capacity limits
    * Review error message for specific validation failures
  </Accordion>

  <Accordion title="Job Running Too Long">
    * Check if job is actually stuck or just slow
    * Review GET\_RUN output for progress
    * Use KILL\_RUN to terminate if truly stuck
    * Optimize prompt to break into smaller tasks
    * Consider increasing timeout if legitimate long runtime
    * Add progress logging to identify bottlenecks
  </Accordion>

  <Accordion title="History Not Showing Recent Runs">
    * Allow time for execution to complete
    * Check if job actually executed (verify next\_run\_time updated)
    * Verify run wasn't cleared by CLEAR\_HISTORY
    * Check offset/limit parameters for pagination
    * Review retention policy settings
    * Query with larger limit to see all history
  </Accordion>
</AccordionGroup>

## **Scheduler Architecture**

Understanding the scheduler components:

```mermaid theme={null}
graph TB
    A[Scheduler Engine] -->|Monitors| B[Job Store]
    B -->|Triggers| C[Executor]
    C -->|Executes| D[Bot Instance]
    D -->|Returns| E[Job Output]
    E -->|Stores| F[Run History]
    A -->|Checks| G[Trigger Conditions]
    G -->|Time-based| H[Cron/Interval/Date]
    F -->|Queries| I[GET_HISTORY API]
    B -->|Manages| J[Job Lifecycle]
```

### **Key Components**

* **Scheduler Engine**: Core APScheduler managing job lifecycle and execution timing
* **Job Store**: Persistent storage for job definitions and configurations
* **Executor**: Thread pool that runs scheduled jobs
* **Trigger Conditions**: Cron, interval, or date-based scheduling logic
* **Run History**: Database storing execution records, outputs, and status
* **Bot Instance**: Target bot that executes the what\_to\_do\_prompt

## **Performance Considerations**

<CardGroup cols={2}>
  <Card title="Avoid Over-Scheduling" icon="gauge">
    Too many frequent jobs can overwhelm system. Use appropriate intervals based on actual needs.
  </Card>

  <Card title="Optimize Prompts" icon="bolt">
    Keep prompts focused and efficient. Break complex operations into separate jobs if needed.
  </Card>

  <Card title="Monitor Resource Usage" icon="server">
    Track job execution times and resource consumption. Optimize or reschedule resource-intensive jobs.
  </Card>

  <Card title="Clean History Regularly" icon="broom">
    Use CLEAR\_HISTORY periodically to prevent history table bloat and maintain performance.
  </Card>
</CardGroup>

## **Comparison: Scheduling Options**

| Feature | Process Scheduler | Airflow | Cron (System) |
| - | - | - | - |
| **Setup Complexity** | ✅ Simple | ⚠️ Complex | ✅ Simple |
| **Scheduling Options** | ✅ Cron, interval, date | ✅ Cron-based | ✅ Cron only |
| **Job Dependencies** | ⚠️ In prompt logic | ✅ Native DAG support | ❌ Manual |
| **Execution History** | ✅ Built-in database | ✅ Metadata database | ⚠️ Logs only |
| **UI Management** | ✅ API-based | ✅ Web UI | ⚠️ Config files |
| **Monitoring** | ✅ GET\_HISTORY API | ✅ Rich monitoring | ⚠️ Manual log checking |
| **Error Handling** | ✅ Captured in history | ✅ Retry logic | ⚠️ Manual |
| **Use Case** | Simple scheduled tasks | Complex workflows | System maintenance |

<Tip>
  Use **process\_scheduler\_tools** for straightforward scheduled tasks within Genesis. For complex multi-step workflows with dependencies, consider **airflow\_tools**. For simple system maintenance, native cron may suffice.
</Tip>

## **Migration Guide**

### **From Cron to Process Scheduler**

```bash theme={null}
# Old cron entry
0 9 * * * /path/to/script.sh

# New scheduled job
scheduler_action(
    action="ADD_JOB",
    job_id="daily_task",
    what_to_do_prompt="Execute the daily task script",
    trigger={"type": "cron", "hour": 9, "minute": 0}
)
```

### **Benefits of Migration**

* ✅ Execution history and monitoring
* ✅ Error handling and retry logic
* ✅ Easy modification without server access
* ✅ Centralized job management
* ✅ No need for script files on server