Block retry_policy argument for StubOperator

providers/standard/src/airflow/providers/standard/decorators/stub.py

@@ -46,6 +46,16 @@ def __init__(
             task_id=task_id,
             **kwargs,
         )
+        # A retry_policy is user Python evaluated in-process by the task runner. Stub tasks
+        # execute on a remote/native worker via the Task Execution Interface and never run the
+        # Python task runner, so the policy would silently never fire. Reject it up front.
+        # (retries is fine -- the server computes retry eligibility regardless of runtime.)
+        if getattr(self, "retry_policy", None) is not None:
+            raise ValueError(
+                "@task.stub does not support `retry_policy`: it runs Python in-process, but stub "
+                "tasks execute on a lang-sdk runtime and never evaluate the policy. Use `retries` "
+                "instead."
+            )
         # Validate python callable
         module = ast.parse(self.get_python_source())
 

providers/standard/tests/unit/standard/decorators/test_stub.py

@@ -22,6 +22,8 @@
 
 from airflow.providers.standard.decorators.stub import stub
 
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
+
 
 def fn_ellipsis(): ...
 
@@ -55,3 +57,15 @@ def fn_code():
 def test_stub_signature(fn, error):
     with error:
         stub(fn)()
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_3_PLUS, reason="retry_policy added in Airflow 3.3")
+def test_stub_rejects_retry_policy():
+    from airflow.sdk.definitions.retry_policy import ExceptionRetryPolicy
+
+    with pytest.raises(ValueError, match="does not support `retry_policy`"):
+        stub(fn_pass, retry_policy=ExceptionRetryPolicy(rules=[]))()
+
+
+def test_stub_allows_retries():
+    stub(fn_pass, retries=5)()