airflow-workflow

---
name: airflow-workflow
description: Execution guide for Airflow scheduled jobs — troubleshooting, updating, conn_id conventions, and cron references
license: Apache-2.0
---

# Airflow Workflow

Execution guide for the scheduler subagent working with Airflow.

## Troubleshoot a Failed Job

1. **Check job status** — `get_scheduler_job(job_id)`
2. **List recent runs** — `list_job_runs(job_id, limit=5)` to find the failed run
3. **Get error log** — `get_run_log(job_id, run_id)` for the failed run_id
4. **Analyze the error** — common failure categories:
   - SQL syntax error → fix SQL and `update_job()`
   - Connection failure → check conn_id in Airflow Connections (Admin > Connections), verify host is reachable from the scheduler worker
   - Timeout → optimize the query or increase resources
   - Permission denied → verify DB credentials in Airflow Connections (Admin > Connections)
5. **Fix and re-run**:
   - Update SQL: `update_job(job_id, sql_file_path=..., job_name=..., conn_id=...)`
   - Manual trigger to verify: `trigger_scheduler_job(job_id)`
   - Confirm success: `list_job_runs(job_id, limit=1)`

## Update an Existing Job

1. **Check current state** — `get_scheduler_job(job_id)` to see existing config
2. **Pause the job** — `pause_job(job_id)` to prevent runs during update
3. **Write SQL** — use `write_file` or `edit_file` to save the new SQL under
   `jobs/<job_name>.sql`
4. **Update** — `update_job(job_id, sql_file_path=..., job_name=..., conn_id=...)`
5. **Resume** — `resume_job(job_id)` to re-enable scheduling
6. **Do not manually trigger** after a normal create/update unless the user
   explicitly asks for an immediate run. Deterministic validation triggers and
   polls deliverable scheduler jobs after the agent returns the target.

## Delete an Existing Job

1. **Confirm with the user** — deletion is destructive.
2. **Delete** — call `delete_job(job_id)`.
3. **Honor the tool result** — if `delete_job` returns `success=0`, report the
   deletion as failed or incomplete. Do not claim completion or success.
4. **Verify only with direct lookup** — use `get_scheduler_job(job_id)` if you
   need a follow-up check. For Airflow, scheduling deletion is complete when
   the job is not found or is inactive/deleted.
5. **Do not rely on list output** — `list_scheduler_jobs` may omit an Airflow DAG
   after its file is removed even while Airflow metadata still exists and blocks
   re-creation with the same job id.
6. **Use precise wording for partial cleanup** — if metadata still exists but
   the DAG is inactive/deleted, say scheduling has been removed and metadata
   cleanup is pending. The same `dag_id` may not be immediately reusable via
   submit; use update or retry cleanup if needed.
7. **Use explicit file deletion only** — `delete_job` owns Airflow DAG file
   removal. For other files, use a dedicated delete-file tool if one is
   available; otherwise report that file deletion is unavailable. Do not
   overwrite or empty files as a substitute for deletion.

## DB Connection (`conn_id`)

`submit_sql_job` and `update_job` require `conn_id` — the Airflow Connection ID for the target database.
The connection is managed entirely by Airflow (Admin > Connections) and resolved at runtime by the scheduler worker.

Available conn_id values are shown in the `submit_sql_job` and `update_job` tool descriptions (from `scheduler.connections` in agent.yml).

## Naming Conventions

- `job_name`: `<frequency>_<domain>_<description>`, e.g. `daily_sales_summary`, `hourly_order_count`
- SQL file: `jobs/<job_name>.sql`

Before calling `submit_sql_job` or `update_job`, create or update that SQL
file with `write_file` / `edit_file`. Do not ask the user to create the file
when filesystem tools are available.

## Common Cron Expressions

| Schedule | Cron |
|----------|------|
| Every day at 8am | `0 8 * * *` |
| Every hour | `0 * * * *` |
| Every 2 hours | `0 */2 * * *` |
| Monday at 9am | `0 9 * * 1` |
| 1st of month at midnight | `0 0 1 * *` |

## Quick Reference

| Goal | Tool |
|------|------|
| Create SQL file | `write_file(path="jobs/<job_name>.sql", content=...)` |
| Submit SQL job | `submit_sql_job(job_name, sql_file_path, conn_id)` |
| Submit SparkSQL job | `submit_sparksql_job(job_name, sql_file_path)` |
| Check job status | `get_scheduler_job(job_id)` |
| List all jobs | `list_scheduler_jobs(limit=20)` |
| Trigger manual run | `trigger_scheduler_job(job_id)` only when explicitly requested or troubleshooting |
| View run history | `list_job_runs(job_id)` |
| View run log | `get_run_log(job_id, run_id)` |
| Pause / Resume | `pause_job(job_id)` / `resume_job(job_id)` |
| Update job | `update_job(job_id, sql_file_path, job_name, conn_id)` |
| Delete job | `delete_job(job_id)` |