Enhance TaskInstance and PlainXComArg to support post-expansion placeholders

[bramhanandlingala]

Description

This change reintroduces the placeholder upstream map-index resolution behavior from #59691 while avoiding the regression that caused incorrect trigger-rule evaluation in mapped task groups.

The placeholder-to-expanded-instance (-1 → 0) resolution is now applied only where required for XCom resolution. Trigger-rule dependency evaluation continues to use the original map-index resolution behavior, preserving per-index expansion semantics for mapped task groups.

This ensures that downstream placeholder tasks can correctly resolve upstream references after expansion without affecting the behavior validated by the regressions addressed in #50210.

Tests

• test_downstream_placeholder_handles_upstream_post_expansion
• test_one_failed_trigger_rule_in_mapped_task_group_is_per_index
• test_one_failed_trigger_rule_runs_on_indirect_failure_in_mapped_task_group

Fixes: #68417

---

Important

🛠️ Maintainer triage note for @bramhanandlingala · by @potiuk · 2026-06-22 06:31 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed (see the Pull Request quality criteria):

• ❌ Pre-commit / static checks — failing: CI image checks / Static checks.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

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

[SameerMesiah97]

I think there is a conceptual issue with this PR. I understand the original issue mentions that my original fix in PR #59691 was addressing XCom Arg resolution but that is not entirely accurate (keep in mind that the body of the issue #68454 is AI-generated so it could be a hallucination). PR #59691 is meant to address an unexpected behaviour during trigger rule evaluation due to a missing upstream placeholder/summary task instance. In other words, XCom Arg resolution is not being touched (that is the precursor step to trigger rule evaluation which is where this the bug addressed in PR #59691 arose).

I have checked one potential fix locally which builds upon PR #67684 by adding TR.NONE_FAILED_MIN_ONE_SUCCESS like this:

is_fast_triggered = task.trigger_rule in (TR.ONE_SUCCESS, TR.ONE_FAILED, TR.ONE_DONE, TR.NONE_FAILED_MIN_ONE_SUCCESS)

I have not run the entire test suite for airflow core but this did not break the existing tests in test_mappedoperator.py and there are other trigger rules under which the bug may arise. But doing the above appeared to have fixed the original issue being addressed by addressed by PR #59691. So you may explore this approach instead.

[SameerMesiah97]

@bramhanandlingala

Do you plan on continuing work on this?

[bramhanandlingala]

Hi @kaxil , @amoghrajesh

All checks are passing and the branch has been updated. Could you please take a look at this PR?
Thanks!

[SameerMesiah97]

Hi @kaxil , @amoghrajesh

All checks are passing and the branch has been updated. Could you please take a look at this PR? Thanks!

@bramhanandlingala

Can you run the following DAG using the code from main with and without your fix and confirm that your fix is giving the intended result:

from datetime import datetime

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.decorators import dag, task, task_group
from airflow.utils.trigger_rule import TriggerRule


@task
def get_the_list():
    return ["one", "two", "three"]


@task
def process_something(x: str):
    output = f"{x}"
    print(output)
    return output


@task_group
def the_task_group(x: str):
    start = EmptyOperator(task_id="start")
    something = process_something(x)
    end = EmptyOperator(task_id="end")

    start >> something >> end


@dag(
    default_args={
        "start_date": datetime(year=2025, month=12, day=1),
        "trigger_rule": TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    }
)
def dtm_bug_reproduce_dag_original_case():
    end = EmptyOperator(task_id="end")

    the_list = get_the_list()

    mapped_group = (
        the_task_group
        .expand(x=the_list)
    )

    the_list >> mapped_group >> end


dag = dtm_bug_reproduce_dag_original_case()

Please provide screenshots if possible.

[bramhanandlingala]

@SameerMesiah97

Thanks for the suggestion. I'll run the provided DAG on the latest main branch and compare the behavior with my fix. I'll report back with the results shortly.