Decouple dag.test from importing from airflow-core

[amoghrajesh]

Body

dag.test today needs airflow-core to work, this should be gone for client server separation: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/definitions/dag.py#L1147-L1376

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[shnhdan]

I'd like to work on this - is it open for external contributors , or is it being handled internally ? Just wanted to know before I start looking into it .

[potiuk]

Just start if you want. If you will have good pr - submit it.

[junyeong0619]

Hi @amoghrajesh,

I got interested in this issue and did some analysis. I'd like to propose a focused PR removing two airflow-core imports inside _run_task (task-sdk/src/airflow/sdk/definitions/dag.py):

1. airflow.utils.session.create_session (L1473) — sub-issue Remove airflow.utils.session references in task-sdk #54710
2. airflow.serialization.serialized_objects.create_scheduler_operator (L1484)

Both are coupled: they exist only because _run_task does an ORM write (session.add(ti)) for the DEFERRED → SCHEDULED transition after inline trigger execution.
Proposal: move the DB-write responsibility from _run_task up to its caller (dag.test), which already has a session in scope. _run_task returns state-transition intent via TaskRunResult; the caller performs the write.

I'd keep the scope limited to _run_task and its single caller in dag.test.

Does this direction look acceptable?