datus-ai/datus-agent/airflow-workflow

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)

1. 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

1. Update — update_job(job_id, sql_file_path=..., job_name=..., conn_id=...)
2. Resume — resume_job(job_id) to re-enable scheduling
3. 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.

1. 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.

1. 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.

1. 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.

1. 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

Quick Reference