Add simple extra links support for TaskFlow tasks

[fredthomsen]

This PR replaces #62079, which was closed and could not be re-opened. The branch has been rebased onto the latest main.

Previously, adding extra links to a TaskFlow task required creating
a custom BaseOperator subclass with operator_extra_links and a
companion BaseOperatorLink class. This is heavyweight for what is
often just a URL set at runtime.

This change lets @task authors declare link names in the decorator
and set URLs directly during execution:

@task(extra_links=["Docs"])
def my_task(extra_links):
    extra_links["Docs"] = "https://example.com"

What's included:

• TaskFlowExtraLink — a concrete BaseOperatorLink that stores
  URLs via in-memory, created at parse time from declared link names
• _build_extra_links() merges TaskFlow links with existing
  class-level links, warning on name conflicts
• ExtraLinksAccessor — dict-like context object injected as
  context["extra_links"] for setting URLs during execution
• Mapped operators retain extra links through unmap()
• Serialization/deserialization works via existing operator_extra_links
  machinery (no new XCom keys or API changes needed)

closes: #54527

Simple DAGs used to verify this solution:

from airflow.decorators import dag, task


@dag
def taskflow_extra_links():
    """
    Vanilla test.
    """
    @task(extra_links=["Docs", "Home"])
    def process_data(extra_links):
        extra_links["Docs"] = "https://airflow.apache.org/docs/"
        extra_links["Home"] = "https://airflow.apache.org/"

    process_data()


taskflow_extra_links()


@dag
def taskflow_extra_links_mapped():
    """
    Task expansion test.
    """
    @task(extra_links=["Map"])
    def process_data(num, extra_links, ti):
        extra_links["Map"] = f"https://example.com/{ti.map_index}/"

    process_data.expand(num=[1,2,3])


taskflow_extra_links_mapped()


@dag
def taskflow_extra_links_undeclared():
    """
    This should blow up.
    """
    @task
    def process_data(extra_links):
        extra_links["Home"] = "invalid"

    process_data()


taskflow_extra_links_undeclared()


@dag
def taskflow_extra_links_virtualenv():
    @task.virtualenv(extra_links=["Venv"])
    def process_data(extra_links):
        extra_links["Venv"] = "https://example.com/virtualenv/"

    process_data()


taskflow_extra_links_virtualenv()

A previous attempt (#55257) stalled because it bypassed the existing
operator_extra_links serialization pipeline: it pushed a plain dict in
finalize() and patched the webserver API with XCom wildcard scans, which
raised concerns about key collisions and split data sources.

This PR sidesteps those issues by creating TaskFlowExtraLink objects at
parse time via @task(extra_links=["Name"]), plugging directly into the
existing operator_extra_links → serialization → XComOperatorLink
pipeline so no webserver changes are needed. URLs are pushed immediately via
SUPERVISOR_COMMS, so @task.virtualenv works transparently. The tradeoff
is that link names must be pre-declared in the decorator rather than created
dynamically at runtime.

---

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

• Yes — Claude Code (Claude Opus 4.6)

Generated-by: Claude Code 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 or {issue_number}.significant.rst, in airflow-core/newsfragments.

---

Important

🛠️ Maintainer triage note for @fredthomsen · by @potiuk · 2026-07-08 15:38 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Provider tests failing (Integration and System Tests / Integration: providers elasticsearch). See the contributor guide.

Full criteria: Pull Request quality criteria.

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