Add support for operators that achieve durable execution manually in registry

[amoghrajesh]

---

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

• Yes: claude sonnet 4.6

---

What

Follow-up to #70289, which added __supports_durable_execution: ClassVar[bool] = True directly to KubernetesPodOperator and AgentOperator both of which implement crash-safe execution against task_state_store without inheriting ResumableJobMixin. That attribute wasn't read by anything yet; the Registry's is_durable_capable() only checked for ResumableJobMixin in the MRO, so both operators stayed invisible to the badge despite genuinely qualifying.

Current behaviour

is_durable_capable() in dev/registry/extract_parameters.py only qualifies a class via the ResumableJobMixin structural check (inherits it, all abstract methods overridden, execute_resumable reachable from execute()). Classes that implement durable execution some other way have no path to qualify.

Proposed change

Adds a second, independent qualifying path: is_durable_capable() now also checks for a class level __supports_durable_execution = True declaration. Because the attribute name uses a double leading underscore, Python name mangles it to _{ClassName}__supports_durable_execution on whichever class declares it. The check computes that mangled name per class being inspected (f"_{cls.__name__}__supports_durable_execution") rather than a fixed string, so it only matches the exact class that wrote the declaration in its own body -- not any subclass that merely inherits from it.

Changes of Note

This distinction matters concretely: KubernetesPodOperator has real subclasses (SparkKubernetesOperator, KubernetesJobOperator, KubernetesStartKueueJobOperator) that override execute() themselves. Nothing verifies those overrides preserve the task_state_store reconnect behavior yet, so the marker must not propagate to them automatically.

Testing

Manual validation after generating registry data shows:

    {
      "id": "common-ai-agent-AgentOperator",
      "name": "AgentOperator",
      "type": "operator",
      "import_path": "airflow.providers.common.ai.operators.agent.AgentOperator",
      "module_path": "airflow.providers.common.ai.operators.agent",
      "short_description": "Run a pydantic-ai Agent with tools and multi-turn reasoning.",
      "docs_url": "https://airflow.apache.org/docs/apache-airflow-providers-common-ai/stable/_api/airflow/providers/common/ai/operators/agent/index.html#airflow.providers.common.ai.operators.agent.AgentOperator",
      "source_url": "https://github.com/apache/airflow/blob/providers-common-ai/0.7.0/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py#L120",
      "category": "common-ai",
      "provider_id": "common-ai",
      "provider_name": "Common AI",
      "supports_durable_execution": true
    },

And

    {
      "id": "cncf-kubernetes-pod-KubernetesPodOperator",
      "name": "KubernetesPodOperator",
      "type": "operator",
      "import_path": "airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator",
      "module_path": "airflow.providers.cncf.kubernetes.operators.pod",
      "short_description": "Execute a task in a Kubernetes Pod.",
      "docs_url": "https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/_api/airflow/providers/cncf/kubernetes/operators/pod/index.html#airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator",
      "source_url": "https://github.com/apache/airflow/blob/providers-cncf-kubernetes/10.20.0/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py#L142",
      "category": "kubernetes",
      "provider_id": "cncf-kubernetes",
      "provider_name": "Kubernetes",
      "supports_durable_execution": true
    },

---

• 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, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[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