fix(tasks): protect bound task state from callable metadata

- prevent callable __dict__ from overriding task identity and broker state
- preserve non-conflicting custom metadata for backward compatibility
- cover callable-object and legacy decorator metadata collisions

Author: GefMar

taskiq/abc/broker.py

@@ -3,7 +3,7 @@
 import warnings
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from collections.abc import AsyncGenerator, Awaitable, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping
 from functools import WRAPPER_ASSIGNMENTS, update_wrapper
 from logging import getLogger
 from typing import (
@@ -49,12 +49,43 @@
 _T = TypeVar("_T")
 _FuncParams = ParamSpec("_FuncParams")
 _ReturnType = TypeVar("_ReturnType")
+_DecoratedTask = TypeVar(
+    "_DecoratedTask",
+    bound=AsyncTaskiqDecoratedTask[Any, Any],
+)
 
 EventHandler: TypeAlias = Callable[[TaskiqState], Awaitable[None] | None]
 
 logger = getLogger("taskiq")
 
 
+def _copy_callable_metadata(
+    task: _DecoratedTask,
+    source: Callable[..., Any],
+) -> _DecoratedTask:
+    """Preserve callable metadata while keeping task-owned state authoritative."""
+    assigned_metadata = tuple(
+        attribute for attribute in WRAPPER_ASSIGNMENTS if hasattr(source, attribute)
+    )
+    update_wrapper(
+        task,
+        source,
+        assigned=assigned_metadata,
+        updated=(),
+    )
+
+    protected_attributes = set(vars(task))
+    for task_type in type(task).__mro__:
+        protected_attributes.update(vars(task_type))
+
+    source_attributes = getattr(source, "__dict__", None)
+    if isinstance(source_attributes, Mapping):
+        for attribute, value in source_attributes.items():
+            if attribute not in protected_attributes:
+                setattr(task, attribute, value)
+    return task
+
+
 def default_id_generator() -> str:
     """
     Default task_id generator.
@@ -541,12 +572,7 @@ def _decorate_task(
         if "return" in sign:
             return_type = sign["return"]
 
-        assigned_metadata = tuple(
-            attribute
-            for attribute in WRAPPER_ASSIGNMENTS
-            if hasattr(metadata_source, attribute)
-        )
-        decorated_task = update_wrapper(
+        decorated_task = _copy_callable_metadata(
             task_cls(
                 broker=self,
                 original_func=func,
@@ -555,7 +581,6 @@ def _decorate_task(
                 return_type=return_type,  # type: ignore
             ),
             metadata_source,
-            assigned=assigned_metadata,
         )
 
         if register:

tests/abc/test_broker.py

@@ -68,6 +68,38 @@ async def test_func() -> None:
     }
 
 
+def test_decorator_metadata_cannot_replace_task_contract() -> None:
+    """Task fields must win over colliding wrapped-function attributes."""
+    broker = _TestBroker()
+
+    def source(value: int) -> int:
+        return value + 1
+
+    vars(source).update(
+        broker="source-broker",
+        custom_metadata="preserved",
+        kiq="source-kiq",
+        labels={"source": "function"},
+        original_func="source-original-func",
+        return_type=str,
+        task_name="source.task-name",
+    )
+
+    registered = broker.task(
+        task_name="expected.task-name",
+        queue="expected",
+    )(source)
+
+    assert registered.broker is broker
+    assert registered.task_name == "expected.task-name"
+    assert registered.labels == {"queue": "expected"}
+    assert registered.original_func is source
+    assert registered.return_type is int
+    assert callable(registered.kiq)
+    assert vars(registered)["custom_metadata"] == "preserved"
+    assert registered(2) == 3
+
+
 def test_decorator_generates_unique_lambda_task_names() -> None:
     """Anonymous tasks must not collide in the broker task registry."""
     broker = _TestBroker()

tests/test_task_builder.py

@@ -52,6 +52,23 @@ def __call__(self, value: int) -> int:
         return value + self.increment
 
 
+class CollidingCallable:
+    """Callable whose application state overlaps with the task API."""
+
+    def __init__(self, increment: int) -> None:
+        self.broker = "source-broker"
+        self.increment = increment
+        self.kiq = "source-kiq"
+        self.labels = {"source": "callable-object"}
+        self.original_func = "source-original-func"
+        self.return_type = str
+        self.task_name = "source.task-name"
+
+    def __call__(self, value: int) -> int:
+        """Increment one value without exposing state as task metadata."""
+        return value + self.increment
+
+
 @task_builder("shared.process_pool")
 def shared_process_pool_add(left: int, right: int) -> int:
     """Add values in a spawned worker process."""
@@ -146,6 +163,43 @@ async def test_callable_object_task_definition_can_be_bound_and_sent() -> None:
     assert result.return_value == 5
 
 
+async def test_callable_object_state_cannot_replace_task_contract() -> None:
+    source = CollidingCallable(2)
+    definition = task_builder(
+        "shared.callable-owned-state",
+        queue="expected",
+    )(source)
+    broker = InMemoryBroker(await_inplace=True)
+
+    registered = broker.register_task(definition)
+    sent_task = await registered.kiq(3)
+    result = await sent_task.wait_result()
+    await broker.shutdown()
+
+    assert registered.broker is broker
+    assert registered.task_name == "shared.callable-owned-state"
+    assert registered.labels == {"queue": "expected"}
+    assert registered.return_type is int
+    assert registered.__wrapped__ is source
+    assert registered.original_func(3) == 5
+    assert callable(registered.kiq)
+    assert vars(registered)["increment"] == 2
+    assert broker.local_task_registry[registered.task_name] is registered
+    assert result.return_value == 5
+
+
+def test_task_binding_preserves_non_conflicting_custom_metadata() -> None:
+    def source(value: int) -> int:
+        return value + 1
+
+    vars(source)["custom_metadata"] = "preserved"
+    definition = task_builder("shared.custom-metadata")(source)
+
+    registered = InMemoryBroker().register_task(definition)
+
+    assert vars(registered)["custom_metadata"] == "preserved"
+
+
 def test_callable_object_binding_rejects_ambiguous_redeclaration() -> None:
     task_builder("shared.callable-collision")(CallableIncrement(2))