Fix @asset reserved keys with default value

task-sdk/src/airflow/sdk/definitions/asset/decorators.py

@@ -87,14 +87,15 @@ def _fetch_asset(name: str) -> Asset:
 
         value: Any
         for key, param in inspect.signature(self.python_callable).parameters.items():
-            if param.default is not inspect.Parameter.empty:
-                value = param.default
-            elif key == "self":
+            # ``self``/``context``/``outlet_events`` are reserved keys.
+            if key == "self":
                 value = _fetch_asset(self._definition_name)
             elif key == "context":
                 value = context
             elif key == "outlet_events":
                 value = context["outlet_events"]
+            elif param.default is not inspect.Parameter.empty:
+                value = param.default
             else:
                 value = _fetch_asset(key)
             yield key, value

task-sdk/tests/task_sdk/definitions/test_asset_decorators.py

@@ -449,6 +449,44 @@ def test_determine_kwargs_defaults(
             mock.call.send(GetAssetByName(name="inlet_asset_1")),
         ]
 
+    @pytest.mark.parametrize(
+        ("python_callable", "reserved_key"),
+        [
+            pytest.param(lambda self=None: None, "self", id="self"),
+            pytest.param(lambda context=None: None, "context", id="context"),
+            pytest.param(lambda outlet_events=None: None, "outlet_events", id="outlet_events"),
+        ],
+    )
+    def test_determine_kwargs_injects_reserved_key_with_default(
+        self, mock_supervisor_comms, func_fixer, python_callable, reserved_key
+    ):
+        """A default on a reserved arg (self/context/outlet_events) must not shadow the injected value."""
+        mock_supervisor_comms.send.side_effect = [
+            AssetResult(name="example_asset_func", uri="example_asset_func", group="asset"),
+        ]
+        fixed_callable = func_fixer(python_callable)
+        definition = asset(schedule=None)(fixed_callable)
+        outlet_events = OutletEventAccessors()
+        context = {"outlet_events": outlet_events}
+
+        op = _AssetMainOperator(
+            task_id="example_asset_func",
+            inlets=[],
+            outlets=[definition],
+            python_callable=fixed_callable,
+            definition_name="example_asset_func",
+        )
+
+        injected = op.determine_kwargs(context=context)[reserved_key]
+
+        assert injected is not None
+        if reserved_key == "self":
+            assert isinstance(injected, Asset)
+        elif reserved_key == "context":
+            assert injected is context
+        else:
+            assert injected is outlet_events
+
     def test_from_definition_custom_name(self, mock_supervisor_comms, func_fixer):
         @func_fixer
         def example_asset_func(self):