[sjyangkevin]

closes: #60328

Why

Add a new depends_on_previous_tasks parameter to BaseOperator that allows a task to declare cross-dagrun dependencies on specific tasks from the previous DagRun.

• depends_on_past=True only blocks a task instance until the same task in the previous DagRun has succeeded.
• wait_for_downstream=True extends this to downstream tasks of the same task.

A → B → C → D → E
dag_run_1 fails at task D.
For dag_run_2, task C should not start until C, D, and E from dag_run_1 have all succeeded.

Usage

C = PythonOperator(
    task_id="C",
    python_callable=process_data,
    depends_on_previous_tasks=["C", "D", "E"],
)

In DagRun(N), task C is scheduled only if tasks C, D, and E in DagRun(N-1) all finished with success or skipped. The first DagRun is never blocked (same behavior as depends_on_past). Fully opt-in and backward compatible.

Trade-offs

Mutual exclusivity vs. composition with depends_on_past

depends_on_previous_tasks is mutually exclusive with depends_on_past and wait_for_downstream. Combining them raises a ValueError at task construction time.

Rationale: depends_on_previous_tasks is a strict superset, to replicate depends_on_past=True, include the task's own task_id in the list. Allowing both simultaneously would create confusing semantics. Failing fast at definition time prevents silent misconfiguration.

Reusing PrevDagrunDep vs. a new dependency class

The implementation adds a branch to the existing PrevDagrunDep._get_dep_statuses() method rather than creating a separate dependency class.

Rationale: both depends_on_past and depends_on_previous_tasks share the same lifecycle; look up the previous DagRun, check task states, push the PAST_DEPENDS_MET xcom. Sharing the same dep class avoids duplicating the DagRun-lookup logic and ensures the existing ignore_depends_on_past flag and DepContext integration work for both paths without additional wiring.

for loop for per-task check

The loop might perform up to 2N queries (_has_tis + _count_unsuccessful_tis per task ID). It gives per-task error messages (user knows exactly which task failed), the list is assumed to be a reasonable small size. However, if this assumption doesn't hold, will look into a better way for it.

Dag Test

The first dag run has no dependencies on previous dag run, and manually mark previous task_c as failed and trigger the next dag run. The current task_c is in running state as one of the previous task_c, task_d, and task_e are not in success state.

Manually mark previous task_d, and then previous task_e as success, as previous task_c still in failed state, the current run is still in running state.

Manually mark all previous task_c, task_d, and task_e as success, the current dag run proceed, and all tasks success.

Example Dag code

from __future__ import annotations

import logging
from datetime import datetime

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

log = logging.getLogger(__name__)


def _log_context(task_id: str, **context):
    """Log the current run's logical date for visibility in task logs."""
    logical_date = context.get("logical_date") or context.get("execution_date")
    log.info("[%s] Running in dag_run with logical_date=%s", task_id, logical_date)


with DAG(
    dag_id="example_depends_on_previous_tasks",
    description=(
        "Demonstrates depends_on_previous_tasks: task C waits for C, D, and E "
        "from the previous DAG run before scheduling."
    ),
    start_date=datetime(2025, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["example", "depends_on_previous_tasks"],
) as dag:
    task_a = PythonOperator(
        task_id="task_a",
        python_callable=_log_context,
        op_kwargs={"task_id": "task_a"},
    )

    task_b = PythonOperator(
        task_id="task_b",
        python_callable=_log_context,
        op_kwargs={"task_id": "task_b"},
    )

    task_c = PythonOperator(
        task_id="task_c",
        python_callable=_log_context,
        op_kwargs={"task_id": "task_c"},
        depends_on_previous_tasks=["task_c", "task_d", "task_e"],
        # depends_on_previous_tasks=["task_a", "task_b"],
        # depends_on_past=True,
    )

    task_d = PythonOperator(
        task_id="task_d",
        python_callable=_log_context,
        op_kwargs={"task_id": "task_d"},
    )

    task_e = PythonOperator(
        task_id="task_e",
        python_callable=_log_context,
        op_kwargs={"task_id": "task_e"},
    )

    # Pipeline topology: A → B → C → D → E
    task_a >> task_b >> task_c >> task_d >> task_e

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [Antigravity] following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

---

Important

🛠️ Maintainer triage note for @sjyangkevin · by @potiuk · 2026-06-18 13:57 UTC

Paused pending your next update — this PR has been inactive for ~50 days, so it's been moved to draft to keep the review queue clear:

• Rebase on the latest main, address any new failures, and mark it Ready for review when you pick it back up — no rush.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[Copilot]

Pull request overview

This PR introduces a new cross-DagRun dependency mechanism, allowing a task to wait on a specific set of task IDs from the previous DagRun (as an opt-in alternative to depends_on_past / wait_for_downstream). It extends operator config, dependency evaluation (PrevDagrunDep), serialization, API schemas, and documentation to support and expose the new field.

Changes:

• Add depends_on_previous_tasks to BaseOperator / mapped operators with mutual-exclusivity validation vs depends_on_past and wait_for_downstream.
• Extend PrevDagrunDep to enforce the previous-DagRun task list success/skipped semantics.
• Plumb the new field through serialization + REST API/OpenAPI models and update docs/tests accordingly.

Reviewed changes

Copilot reviewed 21 out of 21 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
task-sdk/src/airflow/sdk/definitions/mappedoperator.py	Adds getter/setter so mapped tasks can carry depends_on_previous_tasks.
task-sdk/src/airflow/sdk/bases/operator.py	Introduces the new parameter, stores it, and enforces mutual exclusivity.
docs/spelling_wordlist.txt	Adds words used by new docs/description (“retryable”, “superset”).
airflow-ctl/src/airflowctl/api/datamodels/generated.py	Updates CLI API datamodel to include depends_on_previous_tasks.
airflow-core/tests/unit/ti_deps/deps/test_prev_dagrun_dep.py	Adds unit tests for PrevDagrunDep behavior and operator validation.
airflow-core/tests/unit/serialization/test_dag_serialization.py	Updates serialization field expectations to include the new attribute.
airflow-core/tests/unit/models/test_backfill.py	Updates backfill validation tests for the new constraint.
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_tasks.py	Asserts tasks endpoint includes depends_on_previous_tasks in responses.
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py	Updates REST error-message assertions for backfill validation.
airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts	Updates generated UI TS client types with the new field.
airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts	Updates generated UI TS client schema with the new field + required list.
airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py	Implements dependency checks against specific previous-run task IDs.
airflow-core/src/airflow/serialization/schema.json	Adds depends_on_previous_tasks to the serialized operator JSON schema.
airflow-core/src/airflow/serialization/definitions/mappedoperator.py	Exposes depends_on_previous_tasks for serialized mapped operators.
airflow-core/src/airflow/serialization/definitions/baseoperator.py	Adds the field to serialized base operator + serialized field list.
airflow-core/src/airflow/models/dagrun.py	Ensures scheduling behavior accounts for the new cross-run dependency flag.
airflow-core/src/airflow/models/backfill.py	Extends backfill param validation to treat depends_on_previous_tasks like depends_on_past.
airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml	Updates OpenAPI spec to include depends_on_previous_tasks in TaskResponse.
airflow-core/src/airflow/api_fastapi/core_api/datamodels/tasks.py	Adds depends_on_previous_tasks to the FastAPI TaskResponse model.
airflow-core/docs/core-concepts/dags.rst	Documents the new feature and its intended semantics/constraints.
airflow-core/docs/core-concepts/dag-run.rst	Notes catchup sequencing implications when the new parameter is used.

[sjyangkevin]

Thanks, I will check and resolve the copilot comments by this week.

[potiuk]

@sjyangkevin Converting to draft — this PR doesn't yet meet our Pull Request quality criteria.

• ❌ Unresolved review comments (4 threads): please walk through each unresolved review thread. Even if a suggestion looks incorrect or irrelevant — and some of them will be, especially any comments left by automated reviewers like GitHub Copilot — it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. Leaving threads unaddressed for weeks blocks the PR from moving forward.

See the linked criteria for how to fix each item, then mark the PR "Ready for review". This is not a rejection — just an invitation to bring the PR up to standard. No rush.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[potiuk]

Quick follow-up to the triage comment above — one clarification on the "Unresolved review comments" item:

Once you believe a thread has been addressed — whether by pushing a fix, or by replying in-thread with an explanation of why the suggestion doesn't apply — please mark the thread as resolved yourself by clicking the "Resolve conversation" button at the bottom of each thread. Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as "still waiting on the author" and keeps the PR from moving forward. The author doing the resolve-click is the expected convention on this project.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[sjyangkevin]

Thanks for the review and feedback. Sorry, it took a while for me to get back to this PR due to limited bandwidth in the past couple of weeks. I will revisit the project convention to make sure i didn't miss any important updates.

Please let me know if there is further feedback.

[sjyangkevin]

Rebased the branch to main, and update description with test evidence. Would appreciate a review or any feedback. Thanks!

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.