Surface durable execution badge in Airflow Registry (#69651)

* Stage 1: Durable execution detection

* Stage 2: Adding supports_durable_execution to schema

* Stage 3: Wiring up with eleventy to show it on the UI

dev/registry/extract_parameters.py

@@ -88,6 +88,7 @@ class Module:
     category: str
     provider_id: str
     provider_name: str
+    supports_durable_execution: bool
 
 
 def get_category(integration_name: str) -> str:
@@ -373,16 +374,51 @@ def _get_source_line(cls: type) -> int | None:
         return None
 
 
+def load_resumable_job_mixin() -> type | None:
+    """Import ResumableJobMixin for durable-execution capability checks, or None if unavailable."""
+    try:
+        from airflow.sdk import ResumableJobMixin
+
+        return ResumableJobMixin
+    except ImportError:
+        log.warning("Could not import ResumableJobMixin")
+        return None
+
+
+def is_durable_capable(cls: type, resumable_mixin: type | None) -> bool:
+    """Return True if a class fully implements ResumableJobMixin's crash-recovery contract.
+
+    Inheriting the mixin is not sufficient: a complete override is inert unless
+    execute() actually calls execute_resumable().
+    """
+    if resumable_mixin is None or resumable_mixin not in cls.__mro__:
+        return False
+
+    if inspect.isabstract(cls):
+        return False
+
+    execute = getattr(cls, "execute", None)
+    if execute is None:
+        return False
+    try:
+        source = inspect.getsource(execute)
+    except (OSError, TypeError):
+        return False
+
+    return "execute_resumable" in source
+
+
 def discover_classes_from_provider(
     provider_yaml_path: Path,
     base_classes: dict[str, type],
+    resumable_mixin: type | None = None,
     inventory: dict[str, str] | None = None,
     version: str = "",
 ) -> list[dict]:
     """Discover classes from a single provider by importing its modules at runtime.
 
     Reads the provider.yaml to find which modules/classes to inspect, imports them,
-    and returns metadata for each discovered class with all 11 Module fields.
+    and returns metadata for each discovered class with all 12 Module fields.
     """
     with open(provider_yaml_path) as f:
         provider_yaml = yaml.safe_load(f)
@@ -431,7 +467,7 @@ def discover_classes_from_provider(
         category: str = "",
         transfer_desc: str | None = None,
     ) -> dict:
-        """Build a full module entry dict with all 11 fields."""
+        """Build a full module entry dict with all 12 fields."""
         module_name = module_path.split(".")[-1]
         docstring = _get_first_docstring_line(cls_or_obj)
         short_desc = docstring or transfer_desc or f"{integration} {module_type}".strip()
@@ -448,6 +484,7 @@ def discover_classes_from_provider(
             "category": category or get_category(integration),
             "provider_id": provider_id,
             "provider_name": provider_name,
+            "supports_durable_execution": is_durable_capable(cls_or_obj, resumable_mixin),
         }
 
     discovered: list[dict] = []
@@ -889,6 +926,8 @@ def _main_discover(
     base_classes = load_base_classes()
     print(f"Loaded {len(base_classes)} base classes: {', '.join(sorted(base_classes))}")
 
+    resumable_mixin = load_resumable_job_mixin()
+
     # Load all provider.yaml data and map provider_id -> yaml dict / path
     provider_yamls_by_id: dict[str, dict] = {}
     provider_paths_by_id: dict[str, Path] = {}
@@ -923,6 +962,7 @@ def _main_discover(
         discovered = discover_classes_from_provider(
             yaml_path,
             base_classes,
+            resumable_mixin,
             inventory=inventories.get(pid),
             version=version,
         )

dev/registry/registry_contract_models.py

@@ -110,6 +110,7 @@ class ModuleContract(BaseModel):
     category: str
     provider_id: str | None = None
     provider_name: str | None = None
+    supports_durable_execution: bool = False
 
 
 class ModulesCatalogContract(BaseModel):

dev/registry/tests/test_extract_parameters.py

@@ -18,11 +18,14 @@
 
 from __future__ import annotations
 
+import abc
+import builtins
 import json
 import types
 from unittest.mock import patch
 
 import pytest
+import yaml
 from extract_parameters import (
     Module,
     _get_source_line,
@@ -31,6 +34,8 @@ from extract_parameters import (
     compare_with_ast,
     discover_classes_from_provider,
     get_category,
+    is_durable_capable,
+    load_resumable_job_mixin,
 )
 
 
@@ -101,11 +106,147 @@ class TestGetSourceLine:
         assert _get_source_line(DynamicClass) is None
 
 
+# ---------------------------------------------------------------------------
+# load_resumable_job_mixin
+# ---------------------------------------------------------------------------
+class TestLoadResumableJobMixin:
+    def test_returns_none_when_airflow_sdk_unavailable(self):
+        real_import = builtins.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name == "airflow.sdk":
+                raise ImportError("no airflow.sdk here")
+            return real_import(name, *args, **kwargs)
+
+        with patch("builtins.__import__", side_effect=fake_import):
+            assert load_resumable_job_mixin() is None
+
+
+# ---------------------------------------------------------------------------
+# is_durable_capable
+# ---------------------------------------------------------------------------
+class FakeResumableJobMixin(abc.ABC):
+    """Stand-in for airflow.sdk.ResumableJobMixin's abstract-method contract."""
+
+    @abc.abstractmethod
+    def submit_job(self, context):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get_job_status(self, external_id, context):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_job_active(self, status):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def is_job_succeeded(self, status):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def poll_until_complete(self, external_id, context):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get_job_result(self, external_id, context):
+        raise NotImplementedError
+
+    def execute_resumable(self, context):
+        raise NotImplementedError
+
+
+class FullyImplementedResumableOperator(FakeResumableJobMixin):
+    def execute(self, context):
+        return self.execute_resumable(context)
+
+    def submit_job(self, context):
+        return "job-1"
+
+    def get_job_status(self, external_id, context):
+        return "RUNNING"
+
+    def is_job_active(self, status):
+        return status == "RUNNING"
+
+    def is_job_succeeded(self, status):
+        return status == "SUCCEEDED"
+
+    def poll_until_complete(self, external_id, context):
+        return None
+
+    def get_job_result(self, external_id, context):
+        return None
+
+
+class PartiallyImplementedResumableOperator(FakeResumableJobMixin):
+    """Retry-path methods left unoverridden -- would only blow up on an actual crash-recovery retry."""
+
+    def execute(self, context):
+        return self.execute_resumable(context)
+
+    def submit_job(self, context):
+        return "job-1"
+
+    def poll_until_complete(self, external_id, context):
+        return None
+
+    def get_job_result(self, external_id, context):
+        return None
+
+
+class UnwiredResumableOperator(FakeResumableJobMixin):
+    """execute() never calls execute_resumable -- dead capability, never exercised."""
+
+    def execute(self, context):
+        return self.submit_job(context)
+
+    def submit_job(self, context):
+        return "job-1"
+
+    def get_job_status(self, external_id, context):
+        return "RUNNING"
+
+    def is_job_active(self, status):
+        return status == "RUNNING"
+
+    def is_job_succeeded(self, status):
+        return status == "SUCCEEDED"
+
+    def poll_until_complete(self, external_id, context):
+        return None
+
+    def get_job_result(self, external_id, context):
+        return None
+
+
+class PlainOperator:
+    def execute(self, context):
+        return None
+
+
+class TestIsDurableCapable:
+    def test_fully_implemented_and_wired_qualifies(self):
+        assert is_durable_capable(FullyImplementedResumableOperator, FakeResumableJobMixin) is True
+
+    def test_missing_retry_path_overrides_disqualifies(self):
+        assert is_durable_capable(PartiallyImplementedResumableOperator, FakeResumableJobMixin) is False
+
+    def test_implemented_but_not_called_from_execute_disqualifies(self):
+        assert is_durable_capable(UnwiredResumableOperator, FakeResumableJobMixin) is False
+
+    def test_no_mixin_in_mro_disqualifies(self):
+        assert is_durable_capable(PlainOperator, FakeResumableJobMixin) is False
+
+    def test_mixin_unavailable_disqualifies(self):
+        assert is_durable_capable(FullyImplementedResumableOperator, None) is False
+
+
 # ---------------------------------------------------------------------------
 # Module dataclass
 # ---------------------------------------------------------------------------
 class TestModuleDataclass:
-    def test_has_all_11_fields(self):
+    def test_has_all_12_fields(self):
         m = Module(
             id="amazon-s3-S3Hook",
             name="S3Hook",
@@ -118,6 +259,7 @@ class TestModuleDataclass:
             category="amazon-s3",
             provider_id="amazon",
             provider_name="Amazon",
+            supports_durable_execution=False,
         )
         assert m.id == "amazon-s3-S3Hook"
         assert m.provider_name == "Amazon"
@@ -534,6 +676,85 @@ class TestSensorNotClassifiedAsOperator:
         assert result[0]["name"] == "MySensor"
 
 
+# ---------------------------------------------------------------------------
+# TestDiscoverClassesFromProvider: supports_durable_execution wiring
+# ---------------------------------------------------------------------------
+class TestDiscoverClassesFromProviderDurableExecution:
+    def test_marks_durable_capable_and_plain_operators(self, tmp_path):
+        class ResumableOperator(FullyImplementedResumableOperator):
+            __module__ = "airflow.providers.test.operators.spark"
+
+        class PlainProviderOperator(PlainOperator):
+            __module__ = "airflow.providers.test.operators.spark"
+
+        provider_yaml = {
+            "package-name": "apache-airflow-providers-test",
+            "name": "Test",
+            "operators": [
+                {
+                    "integration-name": "Test",
+                    "python-modules": ["airflow.providers.test.operators.spark"],
+                },
+            ],
+        }
+        provider_dir = tmp_path / "test"
+        provider_dir.mkdir()
+        yaml_path = provider_dir / "provider.yaml"
+        yaml_path.write_text(yaml.dump(provider_yaml))
+
+        mod = _make_module(
+            "airflow.providers.test.operators.spark",
+            {
+                "ResumableOperator": ResumableOperator,
+                "PlainProviderOperator": PlainProviderOperator,
+            },
+        )
+
+        with (
+            patch("extract_parameters.PROVIDERS_DIR", tmp_path),
+            patch("extract_parameters.importlib.import_module", return_value=mod),
+        ):
+            result = discover_classes_from_provider(
+                yaml_path, base_classes={}, resumable_mixin=FakeResumableJobMixin
+            )
+
+        by_name = {r["name"]: r for r in result}
+        assert by_name["ResumableOperator"]["supports_durable_execution"] is True
+        assert by_name["PlainProviderOperator"]["supports_durable_execution"] is False
+
+    def test_defaults_to_false_when_mixin_unavailable(self, tmp_path):
+        class ResumableOperator(FullyImplementedResumableOperator):
+            __module__ = "airflow.providers.test.operators.spark"
+
+        provider_yaml = {
+            "package-name": "apache-airflow-providers-test",
+            "name": "Test",
+            "operators": [
+                {
+                    "integration-name": "Test",
+                    "python-modules": ["airflow.providers.test.operators.spark"],
+                },
+            ],
+        }
+        provider_dir = tmp_path / "test"
+        provider_dir.mkdir()
+        yaml_path = provider_dir / "provider.yaml"
+        yaml_path.write_text(yaml.dump(provider_yaml))
+
+        mod = _make_module(
+            "airflow.providers.test.operators.spark",
+            {"ResumableOperator": ResumableOperator},
+        )
+
+        with (
+            patch("extract_parameters.PROVIDERS_DIR", tmp_path),
+            patch("extract_parameters.importlib.import_module", return_value=mod),
+        ):
+            result = discover_classes_from_provider(yaml_path, base_classes={})
+
+        assert result[0]["supports_durable_execution"] is False
+
+
 # ---------------------------------------------------------------------------
 # TestDiscoverClassLevelEntries
 # ---------------------------------------------------------------------------

dev/registry/tests/test_registry_contract_models.py

@@ -22,6 +22,7 @@ import pytest
 from pydantic import ValidationError
 from registry_contract_models import (
     build_openapi_document,
+    validate_modules_catalog,
     validate_provider_parameters,
     validate_provider_version_metadata,
     validate_provider_versions,
@@ -75,6 +76,30 @@ def test_validate_provider_parameters_preserves_mro_alias():
     assert "mro_chain" not in class_entry
 
 
+def _module_payload(**overrides):
+    payload = {
+        "name": "Example",
+        "type": "operator",
+        "import_path": "airflow.providers.test.mod.Example",
+        "short_description": "Example module.",
+        "docs_url": "https://example.invalid/docs",
+        "source_url": "https://example.invalid/source",
+        "category": "test",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_module_contract_accepts_legacy_modules_without_supports_durable_execution():
+    validated = validate_modules_catalog({"modules": [_module_payload()]})
+    assert "supports_durable_execution" not in validated["modules"][0]
+
+
+def test_module_contract_preserves_supports_durable_execution_true():
+    validated = validate_modules_catalog({"modules": [_module_payload(supports_durable_execution=True)]})
+    assert validated["modules"][0]["supports_durable_execution"] is True
+
+
 def test_validate_version_metadata_accepts_legacy_version_modules_without_ids():
     payload = {
         "provider_id": "test",

registry/pnpm-workspace.yaml

@@ -14,11 +14,13 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-
 ---
 packages:
   - '.'
 
+allowBuilds:
+  '@scarf/scarf': false
+
 overrides:
   brace-expansion@<1.1.13: '>=1.1.13'
   liquidjs@<10.25.0: '>=10.25.0'

registry/src/css/main.css

@@ -3252,6 +3252,19 @@ main {
   margin-bottom: var(--space-1);
 }
 
+.provider-detail-page .modules .module .content h3 .durable-badge {
+  display: inline-flex;
+  align-items: center;
+  vertical-align: middle;
+  margin-left: var(--space-2);
+  padding: 0.1rem var(--space-2);
+  font-size: var(--text-xs);
+  font-weight: var(--font-medium);
+  border-radius: var(--radius-lg);
+  background: color-mix(in srgb, var(--color-teal-500) 15%, transparent);
+  color: var(--color-teal-400);
+}
+
 .provider-detail-page .modules .module .content p {
   font-size: var(--text-sm);
   color: var(--text-secondary);

registry/src/provider-version.njk

@@ -355,7 +355,12 @@ eleventyComputed:
             <span>{{ module.type[0] | upper }}</span>
           </div>
           <div class="content">
-            <h3>{{ module.name }}</h3>
+            <h3>
+              {{ module.name }}
+              {% if module.supports_durable_execution %}
+              <span class="durable-badge" title="Reconnects to an already-running job on retry instead of resubmitting. Requires durable=True (default) and depends on operator configuration.">Durable</span>
+              {% endif %}
+            </h3>
             <p>{{ module.short_description }}</p>
             <div class="import-row">
               <code class="path">{{ module.import_path }}</code>