fix: evaluate task annotations consistently across Python versions

CHANGELOG.md

@@ -7,6 +7,10 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 
 ## Unreleased
 
+- [#1025](https://github.com/pytask-dev/pytask/pull/1025) fixes postponed annotation
+  evaluation across Python versions and avoids eager evaluation during task signature
+  inspection. Parameter defaults no longer override module globals or builtins when
+  resolving annotations.
 - [#1022](https://github.com/pytask-dev/pytask/pull/1022) raises the minimum supported
   Python version to 3.11.
 - [#935](https://github.com/pytask-dev/pytask/pull/935) detects when a cached task

src/_pytask/_inspect.py

@@ -1,17 +1,19 @@
 from __future__ import annotations
 
 import ast
+import builtins
 import inspect
 import sys
 from inspect import get_annotations as _get_annotations_from_inspect
+from types import ModuleType
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import cast
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
-__all__ = ["get_annotations"]
+__all__ = ["get_annotations", "get_signature"]
 
 
 def get_annotations(
@@ -31,9 +33,10 @@ def get_annotations(
     * Python 3.14 ships `annotationlib` which exposes the raw annotation source and
       provides the building blocks we reuse here. The module doc explains the available
       formats: https://docs.python.org/3/library/annotationlib.html
-    * Other projects run into the same constraints. Pydantic tracks their work in
-      https://github.com/pydantic/pydantic/issues/12080; we might copy improvements from
-      there once they settle on a stable strategy.
+    * Like Pydantic, we evaluate annotations with separate global and local
+      namespaces and read signatures without evaluating annotations. See
+      https://github.com/pydantic/pydantic/blob/v2.14.1/pydantic/_internal/_typing_extra.py
+      and https://github.com/pydantic/pydantic/issues/12080.
 
     Rationale
     ---------
@@ -48,9 +51,8 @@ def get_annotations(
     [`@task`][pytask.task] at decoration time) we recover the correct per-task values.
     The frame locals capture is essential for
     cases where loop variables are only referenced in annotations (not in the function
-    body or closure). If any of these ingredients are missing—for example on Python
-    versions without `annotationlib` - we fall back to the stdlib implementation,
-    so behaviour on 3.10-3.13 remains unchanged.
+    body or closure). On Python versions without `annotationlib`, we read the raw
+    annotations with `inspect`, then evaluate them using the same namespace logic.
     """
     if not eval_str or not hasattr(obj, "__globals__"):
         return _get_annotations_from_inspect(
@@ -61,25 +63,17 @@ def get_annotations(
         raw_annotations = _get_annotations_from_inspect(
             obj, globals=globals, locals=locals, eval_str=False
         )
-        evaluation_globals = cast(
-            "dict[str, Any]", obj.__globals__ if globals is None else globals
+    else:
+        import annotationlib  # noqa: PLC0415
+
+        raw_annotations = annotationlib.get_annotations(
+            obj, globals=globals, locals=locals, format=annotationlib.Format.STRING
         )
-        evaluation_locals = evaluation_globals if locals is None else locals
-        evaluated_annotations = {}
-        for name, expression in raw_annotations.items():
-            evaluated_annotations[name] = _evaluate_annotation_expression(
-                expression, evaluation_globals, evaluation_locals
-            )
-        return evaluated_annotations
-
-    import annotationlib  # noqa: PLC0415
-
-    raw_annotations = annotationlib.get_annotations(
-        obj, globals=globals, locals=locals, format=annotationlib.Format.STRING
-    )
 
-    evaluation_globals = obj.__globals__ if globals is None else globals
-    evaluation_locals = _build_evaluation_locals(obj, locals)
+    evaluation_globals = cast(
+        "dict[str, Any]", obj.__globals__ if globals is None else globals
+    )
+    evaluation_locals = _build_evaluation_locals(obj, locals, evaluation_globals)
 
     evaluated_annotations = {}
     for name, expression in raw_annotations.items():
@@ -90,14 +84,35 @@ def get_annotations(
     return evaluated_annotations
 
 
+def get_signature(obj: Callable[..., Any]) -> inspect.Signature:
+    """Read a signature without evaluating annotations against unavailable locals.
+
+    Pydantic's `signature_no_eval` uses the same version split. We request STRING
+    instead of FORWARDREF on Python 3.14 so loop-local values are evaluated only
+    when the collection namespace is available.
+    """
+    if sys.version_info >= (3, 14):
+        import annotationlib  # noqa: PLC0415
+
+        return inspect.signature(obj, annotation_format=annotationlib.Format.STRING)
+    return inspect.signature(obj)
+
+
 def _build_evaluation_locals(
-    obj: Callable[..., Any], provided_locals: dict[str, Any] | None
+    obj: Callable[..., Any],
+    provided_locals: dict[str, Any] | None,
+    evaluation_globals: dict[str, Any],
 ) -> dict[str, Any]:
-    # Order matters: later updates override earlier ones.
-    # Default arguments are lowest priority (fallbacks), then provided_locals,
-    # then snapshot_locals (captured loop variables) have highest priority.
-    evaluation_locals: dict[str, Any] = {}
-    evaluation_locals.update(_get_default_argument_locals(obj))
+    evaluation_builtins = evaluation_globals.get("__builtins__", builtins)
+    if isinstance(evaluation_builtins, ModuleType):
+        evaluation_builtins = vars(evaluation_builtins)
+
+    # Recover missing names from defaults without shadowing globals or builtins.
+    evaluation_locals = {
+        name: value
+        for name, value in _get_default_argument_locals(obj).items()
+        if name not in evaluation_globals and name not in evaluation_builtins
+    }
     if provided_locals:
         evaluation_locals.update(provided_locals)
     evaluation_locals.update(_get_snapshot_locals(obj))
@@ -112,7 +127,7 @@ def _get_snapshot_locals(obj: Callable[..., Any]) -> dict[str, Any]:
 
 def _get_default_argument_locals(obj: Callable[..., Any]) -> dict[str, Any]:
     try:
-        parameters = inspect.signature(obj).parameters.values()
+        parameters = get_signature(obj).parameters.values()
     except (TypeError, ValueError):
         return {}
 

src/_pytask/collect_utils.py

@@ -2,14 +2,14 @@
 
 from __future__ import annotations
 
-import inspect
 from dataclasses import replace
 from typing import TYPE_CHECKING
 from typing import Annotated
 from typing import Any
 from typing import get_origin
 
 from _pytask._inspect import get_annotations
+from _pytask._inspect import get_signature
 from _pytask.exceptions import NodeNotCollectedError
 from _pytask.models import NodeInfo
 from _pytask.node_protocols import NodeTree
@@ -187,7 +187,7 @@ def parse_products_from_task_function(  # noqa: C901
     signature_defaults = parse_keyword_arguments_from_signature_defaults(obj)
     kwargs = {**signature_defaults, **task_kwargs}
 
-    parameters = list(inspect.signature(obj).parameters)
+    parameters = list(get_signature(obj).parameters)
     parameters_with_product_annot = _find_args_with_product_annotation(obj)
     parameters_with_node_annot = _find_args_with_node_annotation(obj)
 

src/_pytask/execute_utils.py

@@ -2,10 +2,10 @@
 
 from __future__ import annotations
 
-import inspect
 from typing import TYPE_CHECKING
 from typing import Any
 
+from _pytask._inspect import get_signature
 from _pytask.exceptions import NodeLoadError
 from _pytask.tree_util import tree_map
 
@@ -25,7 +25,7 @@ def _safe_load(node: PNode | PProvisionalNode, task: PTask, *, is_product: bool)
 
 def execute_task(task: PTask) -> Any:
     """Load task arguments and invoke its body once."""
-    parameters = inspect.signature(task.function).parameters
+    parameters = get_signature(task.function).parameters
 
     kwargs = {}
     for name, value in task.depends_on.items():

src/_pytask/task_utils.py

@@ -21,6 +21,7 @@
 from typing import cast
 from typing import overload
 
+from _pytask._inspect import get_signature
 from _pytask.coiled_utils import Function
 from _pytask.coiled_utils import extract_coiled_function_kwargs
 from _pytask.console import get_file
@@ -439,7 +440,7 @@ def parse_keyword_arguments_from_signature_defaults(
     task: Callable[..., Any],
 ) -> dict[str, Any]:
     """Parse keyword arguments from signature defaults."""
-    parameters = inspect.signature(task).parameters
+    parameters = get_signature(task).parameters
     kwargs = {}
     for parameter in parameters.values():
         if parameter.default is not parameter.empty:
@@ -451,7 +452,7 @@ def _generate_ids_for_tasks(
     tasks: list[tuple[str, TaskFunction]],
 ) -> dict[str, TaskFunction]:
     """Generate unique ids for parametrized tasks."""
-    parameters = inspect.signature(tasks[0][1]).parameters
+    parameters = get_signature(tasks[0][1]).parameters
 
     out: dict[str, TaskFunction] = {}
     for i, (name, task) in enumerate(tasks):

tests/test_collect.py

@@ -507,15 +507,18 @@ def task_example() -> Annotated[str, path]:
     assert tmp_path.joinpath("out-1.txt").exists()
 
 
-def test_lazy_annotations_use_current_globals(tmp_path):
-    source = """
+@pytest.mark.parametrize("decorator", ["", "@task", "@task()"])
+def test_lazy_annotations_use_current_globals(tmp_path, decorator):
+    source = f"""
     from __future__ import annotations
 
     from pathlib import Path
     from typing import Annotated
+    from pytask import task
 
     OUTPUT = Path("first.txt")
 
+    {decorator}
     def task_example() -> Annotated[str, OUTPUT]:
         return "Hello"
 
@@ -529,6 +532,52 @@ def task_example() -> Annotated[str, OUTPUT]:
     assert not tmp_path.joinpath("first.txt").exists()
 
 
+def test_annotation_module_globals_take_precedence_over_parameter_defaults(tmp_path):
+    source = """
+    from __future__ import annotations
+    from pathlib import Path
+    from typing import Annotated
+
+    output = Path("out.txt")
+
+    def task_example(output: Path = Path("in.txt")) -> Annotated[str, output]:
+        return output.read_text()
+    """
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+    tmp_path.joinpath("in.txt").write_text("Hello")
+
+    session = build(paths=tmp_path)
+
+    assert session.exit_code == ExitCode.OK
+    assert tmp_path.joinpath("out.txt").read_text() == "Hello"
+    assert tmp_path.joinpath("in.txt").read_text() == "Hello"
+
+
+def test_lazy_annotations_survive_deleted_definition_local(tmp_path):
+    source = """
+    from pathlib import Path
+    from typing import Annotated
+    from pytask import task
+
+    def create_task():
+        output = Path("out.txt")
+
+        @task
+        def task_example() -> Annotated[str, output]:
+            return "Hello"
+
+        del output
+
+    create_task()
+    """
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    session = build(paths=tmp_path)
+
+    assert session.exit_code == ExitCode.OK
+    assert tmp_path.joinpath("out.txt").read_text() == "Hello"
+
+
 def test_string_literal_annotations_are_resolved(tmp_path):
     source = """
     from __future__ import annotations

tests/test_inspect.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Annotated
+
+from _pytask._inspect import get_annotations
+
+
+def test_postponed_annotations_resolve_default_arguments_after_factory_returns():
+    def factory():
+        output = Path("out.txt")
+
+        def example(output: Path = output) -> Annotated[str, output]:
+            return str(output)
+
+        return example
+
+    annotations = get_annotations(factory(), eval_str=True)
+
+    assert annotations == {"output": Path, "return": Annotated[str, Path("out.txt")]}
+
+
+def test_explicit_annotation_namespaces_override_function_globals():
+    def example(value: Path) -> Path:
+        return value
+
+    annotations = get_annotations(
+        example, globals={"Path": int}, locals={"Path": str}, eval_str=True
+    )
+
+    assert annotations == {"value": str, "return": str}
+
+
+def test_parameter_defaults_do_not_override_builtins():
+    def example(str: str = "value") -> str:  # noqa: A002
+        return str
+
+    annotations = get_annotations(example, eval_str=True)
+
+    assert annotations == {"str": str, "return": str}