Expose task state and asset state to tasks via context vars

[amoghrajesh]

Body

Wire task state and asset state into the task execution runtime so tasks can access them via context variable

Committer

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

[Geethapranay1]

hi @amoghrajesh , i like to take up this feature.
I plan to use the pattern from VariableAccessor and ConnectionAccessor. My approach requires three steps.
I will add TaskStateAccessor and AssetStateAccessor inside the context.py file, then I will inject task state and asset state into the template context inside the task_runner.py file and I will wire missing communication endpoints for asset metadata inside the comms.py file and the execution API. can i start to implement now? Please share your thoughts.