fix(collection): validate unique task signatures

CHANGELOG.md

@@ -10,6 +10,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - [#889](https://github.com/pytask-dev/pytask/pull/889) improves typing for tree
   operations by wrapping optree's pytree utilities with pytask-specific signatures
   and requiring optree 0.16.0 or newer.
+- [#932](https://github.com/pytask-dev/pytask/pull/932) validates unique task signatures
+  during collection.
 
 ## 0.6.0 - 2026-05-01
 

src/_pytask/collect.py

@@ -53,6 +53,7 @@
 from _pytask.task_utils import COLLECTED_TASKS
 from _pytask.task_utils import parse_collected_tasks_with_task_marker
 from _pytask.task_utils import task as task_decorator
+from _pytask.task_utils import validate_unique_task_signatures
 from _pytask.typing import TaskFunction
 from _pytask.typing import is_task_function
 
@@ -81,6 +82,7 @@ def pytask_collect(session: Session) -> bool:
 
     try:
         session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks)
+        validate_unique_task_signatures(session.tasks)
     except Exception:  # noqa: BLE001  # pragma: no cover
         report = CollectionReport.from_exception(
             outcome=CollectionOutcome.FAIL, exc_info=sys.exc_info()

src/_pytask/provisional.py

@@ -20,6 +20,7 @@
 from _pytask.reports import ExecutionReport
 from _pytask.task_utils import COLLECTED_TASKS
 from _pytask.task_utils import parse_collected_tasks_with_task_marker
+from _pytask.task_utils import validate_unique_task_signatures
 from _pytask.tree_util import tree_map
 from _pytask.tree_util import tree_map_with_path
 from _pytask.typing import is_task_generator
@@ -103,6 +104,7 @@ def pytask_execute_task(session: Session, task: PTask) -> None:
             session.hook.pytask_collect_modify_tasks(
                 session=session, tasks=session.tasks
             )
+            validate_unique_task_signatures(session.tasks)
             # Append the last collection report after successful modification
             if report:
                 session.collection_reports.append(report)
@@ -111,6 +113,8 @@ def pytask_execute_task(session: Session, task: PTask) -> None:
                 task=task, exc_info=sys.exc_info()
             )
             session.execution_reports.append(exec_report)
+            session.should_stop = True
+            return
 
         recreate_dag(session, task)
 

src/_pytask/task_utils.py

@@ -27,6 +27,8 @@
 from _pytask.mark import Mark
 from _pytask.models import CollectionMetadata
 from _pytask.models import ParsedAfter
+from _pytask.nodes import Task
+from _pytask.nodes import TaskWithoutPath
 from _pytask.shared import find_duplicates
 from _pytask.shared import unwrap_task_function
 from _pytask.typing import TaskFunction
@@ -37,6 +39,8 @@
     from pathlib import Path
     from uuid import UUID
 
+    from _pytask.node_protocols import PTask
+
 P = ParamSpec("P")
 R_co = TypeVar("R_co", covariant=True)
 
@@ -61,6 +65,7 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]:
     "parse_collected_tasks_with_task_marker",
     "parse_keyword_arguments_from_signature_defaults",
     "task",
+    "validate_unique_task_signatures",
 ]
 
 
@@ -74,6 +79,74 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]:
 """
 
 
+def validate_unique_task_signatures(tasks: list[PTask]) -> None:
+    """Raise an error if multiple tasks have the same signature."""
+    signature_to_tasks: dict[str, list[PTask]] = defaultdict(list)
+    for task_ in tasks:
+        signature_to_tasks[task_.signature].append(task_)
+
+    collisions = {
+        signature: tasks_
+        for signature, tasks_ in signature_to_tasks.items()
+        if len(tasks_) > 1
+    }
+    if not collisions:
+        return
+
+    lines = ["Conflicting task identities:"]
+    groups = sorted(
+        collisions.values(), key=lambda group: sorted(task_.name for task_ in group)
+    )
+    for conflicting_tasks in groups:
+        names = ", ".join(
+            repr(name) for name in sorted({t.name for t in conflicting_tasks})
+        )
+        lines.extend(("", f"Tasks sharing an identity: {names}"))
+        lines.extend(f"- {_describe_task(task_)}" for task_ in conflicting_tasks)
+        if all(type(task_) is TaskWithoutPath for task_ in conflicting_tasks):
+            guidance = (
+                "These tasks have the same name. Choose distinct 'name' values "
+                "for TaskWithoutPath tasks."
+            )
+        elif all(type(task_) is Task for task_ in conflicting_tasks):
+            guidance = (
+                "Each Task must have a unique combination of 'path' and 'base_name'. "
+                "Choose distinct 'base_name' values for tasks in the same file; "
+                "changing the display name does not change task identity."
+            )
+        else:
+            guidance = (
+                "These tasks share a signature. For custom task types, ensure the "
+                "'signature' implementation returns a distinct, stable value for "
+                "each task identity."
+            )
+        lines.extend(
+            ("", guidance + " Alternatively, remove the duplicate registration.")
+        )
+
+    raise ValueError("\n".join(lines))
+
+
+def _describe_task(task_: PTask) -> str:
+    """Return a task description which distinguishes conflicting definitions."""
+    function = unwrap_task_function(task_.function)
+    while isinstance(function, functools.partial):
+        function = unwrap_task_function(function.func)
+    callable_name = getattr(function, "__qualname__", type(function).__qualname__)
+    module = getattr(function, "__module__", None)
+    if module:
+        callable_name = f"{module}.{callable_name}"
+
+    try:
+        path = get_file(function)
+        line_number = inspect.getsourcelines(function)[1]
+    except (OSError, TypeError):
+        location = "<unknown>"
+    else:
+        location = f"{path.as_posix()}:{line_number}" if path else "<unknown>"
+    return f"{task_.name!r}: {callable_name} ({location})"
+
+
 @overload
 def task(
     name: Callable[P, R_co],

tests/test_collect.py

@@ -12,17 +12,69 @@
 from _pytask.collect import _find_shortest_uniquely_identifiable_name_for_tasks
 from _pytask.collect import pytask_collect_node
 from _pytask.node_protocols import PPathNode
+from _pytask.pluginmanager import get_plugin_manager
+from _pytask.task_utils import validate_unique_task_signatures
 from pytask import CollectionOutcome
 from pytask import ExitCode
 from pytask import NodeInfo
+from pytask import PathNode
 from pytask import PickleNode
 from pytask import Session
 from pytask import Task
+from pytask import TaskWithoutPath
 from pytask import build
 from pytask import cli
+from pytask import hookimpl
 from tests.conftest import noop
 
 
+def test_duplicate_programmatic_task_signatures_fail_collection(tmp_path):
+    calls = []
+
+    def first():
+        calls.append("first")
+
+    def second():
+        calls.append("second")
+
+    tasks = [
+        TaskWithoutPath(name="duplicate", function=first),
+        TaskWithoutPath(name="duplicate", function=second),
+    ]
+
+    session = build(tasks=tasks, paths=tmp_path, force=True)
+
+    assert session.exit_code == ExitCode.COLLECTION_FAILED
+    assert not session.execution_reports
+    assert not calls
+
+
+def test_duplicate_task_signature_message_contains_all_collision_groups():
+    def first():
+        pass
+
+    def second():
+        pass
+
+    tasks = [
+        TaskWithoutPath(name="duplicate-a", function=first),
+        TaskWithoutPath(name="duplicate-a", function=second),
+        TaskWithoutPath(name="duplicate-b", function=first),
+        TaskWithoutPath(name="duplicate-b", function=second),
+    ]
+
+    with pytest.raises(ValueError, match="Conflicting task identities") as exc_info:
+        validate_unique_task_signatures(tasks)
+
+    message = str(exc_info.value)
+    assert "Conflicting task identities" in message
+    assert "duplicate-a" in message
+    assert "duplicate-b" in message
+    assert all(task.signature not in message for task in tasks)
+    assert "Choose distinct 'name' values" in message
+    assert f"{Path(__file__).as_posix()}:" in message
+
+
 def _make_local_upath_uri(path: Path, protocol: str) -> str:
     return f"{protocol}:///{path.as_posix().lstrip('/')}"
 
@@ -737,3 +789,103 @@ def test_print_warning_if_non_matching_path_is_passed(runner, tmp_path):
     assert result.exit_code == ExitCode.OK
     assert "Collected 0 tasks" in result.output
     assert "Warning: The path" in result.output
+
+
+@pytest.mark.parametrize("introduce_collision", [True, False])
+def test_validate_task_signatures_after_hook_wrapper(
+    tmp_path, monkeypatch, introduce_collision
+):
+    class Plugin:
+        @hookimpl(wrapper=True)
+        def pytask_collect_modify_tasks(self, tasks):
+            result = yield
+            tasks[1].name = "first" if introduce_collision else "second"
+            return result
+
+    pm = get_plugin_manager()
+    pm.register(Plugin())
+    monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm)
+    calls = []
+    tasks = [
+        TaskWithoutPath(name="first", function=lambda: calls.append("first")),
+        TaskWithoutPath(
+            name="second" if introduce_collision else "first",
+            function=lambda: calls.append("second"),
+        ),
+    ]
+
+    session = build(tasks=tasks, paths=tmp_path, force=True)
+
+    if introduce_collision:
+        assert session.exit_code == ExitCode.COLLECTION_FAILED
+        assert not calls
+        assert any(
+            report.exc_info and "Conflicting task identities" in str(report.exc_info[1])
+            for report in session.collection_reports
+        )
+    else:
+        assert session.exit_code == ExitCode.OK
+        assert sorted(calls) == ["first", "second"]
+
+
+def test_duplicate_task_diagnostics_use_callable_source(tmp_path):
+    tasks = [
+        Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop),
+        Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop),
+    ]
+
+    with pytest.raises(ValueError, match="Conflicting task identities") as exc_info:
+        validate_unique_task_signatures(tasks)
+
+    message = str(exc_info.value)
+    assert "tests.conftest.noop" in message
+    assert f"{Path(__file__).with_name('conftest.py').as_posix()}:" in message
+    assert f"({tmp_path.as_posix()}/task_other.py:" not in message
+    assert "remove the duplicate registration" in message
+    assert "unique combination of 'path' and 'base_name'" in message
+    assert "changing the display name does not change task identity" in message
+
+
+def test_duplicate_task_diagnostics_without_source():
+    tasks = [TaskWithoutPath(name="duplicate", function=len) for _ in range(2)]
+
+    with pytest.raises(ValueError, match=r"builtins.len \(<unknown>\)"):
+        validate_unique_task_signatures(tasks)
+
+
+def test_unique_tasks_can_share_dependency(tmp_path):
+    dependency = tmp_path / "input.txt"
+    dependency.write_text("input")
+    calls = []
+    tasks = [
+        TaskWithoutPath(
+            name=name,
+            function=lambda path: calls.append(path.read_text()),
+            depends_on={"path": PathNode(path=dependency)},
+        )
+        for name in ("first", "second")
+    ]
+
+    session = build(tasks=tasks, paths=tmp_path, force=True)
+
+    assert session.exit_code == ExitCode.OK
+    assert calls == ["input", "input"]
+
+
+def test_custom_task_identity_conflict_guidance():
+    class CustomTask(TaskWithoutPath):
+        @property
+        def signature(self):
+            return "shared-custom-signature"
+
+    tasks = [CustomTask(name=name, function=noop) for name in ("first", "second")]
+
+    with pytest.raises(ValueError, match="Conflicting task identities") as exc_info:
+        validate_unique_task_signatures(tasks)
+
+    message = str(exc_info.value)
+    assert "Tasks sharing an identity: 'first', 'second'" in message
+    assert "shared-custom-signature" not in message
+    assert "'signature' implementation" in message
+    assert "distinct, stable value" in message
+    assert "These tasks have the same name" not in message

tests/test_provisional.py

@@ -2,10 +2,14 @@
 
 import textwrap
 
+import pytest
+
+from _pytask.pluginmanager import get_plugin_manager
 from pytask import ExitCode
 from pytask import TaskOutcome
 from pytask import build
 from pytask import cli
+from pytask import hookimpl
 
 
 def test_task_that_produces_provisional_path_node(tmp_path):
@@ -280,3 +284,51 @@ def task_example(
     assert result.exit_code == ExitCode.OK
     assert tmp_path.joinpath("subfolder", "a.txt").exists()
     assert tmp_path.joinpath("subfolder", "b.txt").exists()
+
+
+@pytest.mark.parametrize("collision_in_wrapper", [False, True])
+def test_generated_task_identity_conflict_stops_before_dag_rebuild(
+    tmp_path, monkeypatch, collision_in_wrapper
+):
+    source = """
+    from pathlib import Path
+    from pytask import task, mark
+
+    def task_existing():
+        Path(__file__).with_name("existing-ran").touch()
+
+    @mark.try_first
+    @task(is_generator=True)
+    def task_generator():
+        @task(name=GENERATED_NAME)
+        def generated():
+            Path(__file__).with_name("generated-ran").touch()
+    """
+    generated_name = "task_generated" if collision_in_wrapper else "task_existing"
+    source = source.replace("GENERATED_NAME", repr(generated_name))
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    class Plugin:
+        @hookimpl(wrapper=True)
+        def pytask_collect_modify_tasks(self, tasks):
+            result = yield
+            for task in tasks:
+                if task.base_name == "task_generated":
+                    task.base_name = "task_existing"
+            return result
+
+    pm = get_plugin_manager()
+    pm.register(Plugin())
+    monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm)
+    session = build(paths=tmp_path, force=True)
+
+    assert session.exit_code == ExitCode.FAILED
+    assert session.should_stop
+    assert not tmp_path.joinpath("existing-ran").exists()
+    assert not tmp_path.joinpath("generated-ran").exists()
+    assert any(
+        report.exc_info and "Conflicting task identities" in str(report.exc_info[1])
+        for report in session.execution_reports
+    )
+    assert len(session.tasks) == 3
+    assert len(session.dag.nodes) == 2