fix(providers/standard): improve error message for non-serializable op_kwargs in virtualenv operators

providers/standard/src/airflow/providers/standard/operators/python.py

@@ -579,9 +579,30 @@ def resolve_proxies(obj):
 
         if self.op_args or self.op_kwargs:
             self.log.info("Use %r as serializer.", self.serializer)
-            file.write_bytes(
-                self.pickling_library.dumps({"args": self.op_args, "kwargs": resolve_proxies(self.op_kwargs)})
-            )
+            resolved_kwargs = resolve_proxies(self.op_kwargs)
+            try:
+                file.write_bytes(
+                    self.pickling_library.dumps({"args": self.op_args, "kwargs": resolved_kwargs})
+                )
+            except Exception:
+                # Identify which specific kwarg(s) failed to serialize so the
+                # user gets a clear error instead of an opaque PicklingError.
+                bad_keys: list[str] = []
+                for key, value in resolved_kwargs.items():
+                    try:
+                        self.pickling_library.dumps(value)
+                    except Exception:
+                        bad_keys.append(key)
+                if bad_keys:
+                    raise AirflowException(
+                        f"Failed to serialize op_kwargs. The following keys contain objects that "
+                        f"cannot be pickled: {bad_keys}. This often happens when "
+                        f"render_template_as_native_obj=True and templates like '{{{{ ti }}}}' "
+                        f"resolve to live Airflow objects instead of strings. Use string "
+                        f"representations or pass only the specific attributes you need "
+                        f"(e.g. '{{{{ ti.task_id }}}}', '{{{{ ti.run_id }}}}')."
+                    )
+                raise
 
     def _write_string_args(self, file: Path):
         file.write_text("\n".join(map(str, self.string_args)))

providers/standard/tests/unit/standard/operators/test_python.py

@@ -1092,6 +1092,25 @@ def f():
         op = self.opcls(task_id="task", python_callable=f, **self.default_kwargs())
         copy.deepcopy(op)
 
+    def test_write_args_non_serializable_op_kwargs(self, tmp_path):
+        """Non-serializable op_kwargs should raise AirflowException with helpful message."""
+
+        class NonSerializable:
+            def __reduce__(self):
+                raise TypeError("cannot pickle this")
+
+        def f(x):
+            return x
+
+        op = self.opcls(
+            task_id="task",
+            python_callable=f,
+            op_kwargs={"bad_obj": NonSerializable(), "good_obj": "hello"},
+            **self.default_kwargs(),
+        )
+        with pytest.raises(AirflowException, match=r"cannot be pickled.*\['bad_obj'\]"):
+            op._write_args(tmp_path / "args.pkl")
+
     def test_virtualenv_serializable_context_fields(self, create_task_instance):
         """Ensure all template context fields are listed in the operator.