[kalluripradeep]

Closes #68796

What's the problem?

AirflowRuntimeVaryingValueChecker.visit_With() called exit_dag_context() unconditionally at the end of every with-statement. When a non-DAG with-statement (e.g. with open(...) as f) was nested inside a DAG with-block, it would reset is_in_dag_context=False. Any tasks constructed after the nested with — but still inside the DAG block — were invisible to the checker and never flagged for runtime-varying values even when they used datetime.now(), random(), etc.

with DAG('my_dag') as dag:
    with open('file.txt') as f:   # <-- this incorrectly exited DAG context
        data = f.read()
    # BUG: datetime.now() not flagged — checker thinks we're outside DAG
    t1 = BashOperator(task_id='t', bash_command=str(datetime.now()))

Root cause

# visit_With (before fix) — unconditional, wrong
self.dag_detector.exit_dag_context()

Fix

Guard the call with if is_with_dag_context so only the with-statement that actually entered the DAG context exits it.

if is_with_dag_context:
    self.dag_detector.exit_dag_context()

Changes

• airflow-core/src/airflow/utils/dag_version_inflation_checker.py — one-line guard
• airflow-core/tests/unit/utils/test_dag_version_inflation_checker.py — 4 regression tests

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

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test