fix(wrapper): stop Wrap mutating the ingress it is given (#84)

`preserve_signature` wrote `__signature__`/`__annotations__` onto the ingress object.
A decorator normally defines one ingress and reuses it for every function it wraps, so
that write corrupted the caller's function and made every later Wrap built from the same
ingress advertise the FIRST wrapped function's signature -- while still executing
correctly, so nothing surfaced the lie:

    def shared_ingress(*args, **kwargs): return args, kwargs
    def f(a: int) -> int: ...
    def g(x: str, y: str) -> str: ...

    Wrap(f, ingress=shared_ingress)
    signature(shared_ingress)          # (a: int) -> int   <- caller's function mutated
    Sig(Wrap(g, ingress=shared_ingress))  # (a: int) -> str <- wrong, but g still works

That is the worst shape of bug for this package: silent, and wrong in exactly the
introspection i2 exists to provide (meshed builds DAGs from these signatures).

Preserving means *the wrapper* presents func's interface, so read the signature from
func and never write it onto the ingress. Defaults now come from the same source as the
signature, so the two cannot disagree.

Also, while in here:

- `_get_return_annotation` had `x if x is not Parameter.empty else empty` three times.
  `empty IS Parameter.empty`, so each was a no-op wrapped around a duplicated fallback.
  Reduced to one "egress wins if annotated, else func" rule with doctests, including the
  case that made the dead branch look meaningful (an unannotated func gives `empty`,
  not `None` -- `None` is a real annotation meaning "returns None").
- Extracted `_is_generic_signature`, the actual predicate `'auto'` turns on, with
  doctests for the near-misses ((*a) alone, (x, *a, **kw)).
- Dropped the unused `func` parameter from `_should_preserve_signature`.
- Named the `'auto'` sentinel `AUTO_PRESERVE_SIGNATURE` so it has one definition.
- Converted the numpydoc docstrings to the `:param:` style used everywhere else here.

Public API unchanged. Regression tests fail on the old code and pass on the new.
i2: 470 passed. meshed (biggest dependent): 60 passed. crude/o failures are pre-existing
and byte-identical with and without this change.

Claude-Session: https://claude.ai/code/session_01GsUw8ey8KikzNFQ1UaWWia

Author: thorwhalen

i2/tests/test_wrapper.py

@@ -6,6 +6,7 @@
 import pytest
 
 from i2.wrapper import (
+    Wrap,
     wrap,
     mk_ingress_from_name_mapper,
     rm_params,
@@ -534,3 +535,66 @@ def decorate(func=None, *, multiplier=2):
 
     with pytest.raises(TypeError):
         decorate(_incr, func=_incr)
+
+
+# ---------------------------------------------------------------------------------------
+# Regression: Wrap must not mutate the ingress it is given.
+# `preserve_signature` used to stamp `__signature__`/`__annotations__` onto the ingress
+# object. A decorator normally defines one ingress and reuses it for every function it
+# wraps, so that write corrupted the caller's function AND made every later Wrap built
+# from the same ingress advertise the first-wrapped function's signature -- while still
+# executing correctly, so nothing surfaced the lie.
+
+
+def _shared_ingress(*args, **kwargs):
+    """A module-level ingress, of the shape `preserve_signature='auto'` acts on."""
+    return args, kwargs
+
+
+def test_wrap_does_not_mutate_the_ingress():
+    """Wrapping must leave the caller's ingress object exactly as it was."""
+    from inspect import signature
+
+    def func(a: int) -> int:
+        return a
+
+    before = signature(_shared_ingress)
+    Wrap(func, ingress=_shared_ingress)
+
+    assert signature(_shared_ingress) == before
+    assert not hasattr(_shared_ingress, "__signature__")
+
+
+def test_reused_ingress_gives_each_wrap_its_own_signature():
+    """Two Wraps sharing one ingress must each report their own func's signature."""
+
+    def f(a: int) -> int:
+        return a
+
+    def g(x: str, y: str) -> str:
+        return x + y
+
+    wrapped_f = Wrap(f, ingress=_shared_ingress)
+    wrapped_g = Wrap(g, ingress=_shared_ingress)
+
+    assert str(Sig(wrapped_f)) == "(a: int) -> int"
+    assert str(Sig(wrapped_g)) == "(x: str, y: str) -> str"
+    # ... and both still actually work
+    assert wrapped_f(3) == 3
+    assert wrapped_g("a", "b") == "ab"
+
+
+def test_wrap_signature_is_order_independent():
+    """Building the two Wraps in the other order must give the same signatures."""
+
+    def f(a: int) -> int:
+        return a
+
+    def g(x: str, y: str) -> str:
+        return x + y
+
+    wrapped_g = Wrap(g, ingress=_shared_ingress)
+    wrapped_f = Wrap(f, ingress=_shared_ingress)
+
+    assert str(Sig(wrapped_g)) == "(x: str, y: str) -> str"
+    assert str(Sig(wrapped_f)) == "(a: int) -> int"

i2/wrapper.py

@@ -103,6 +103,11 @@
 InnerKwargs = dict
 KwargsTrans = Callable[[OuterKwargs], InnerKwargs]
 
+#: ``preserve_signature`` value meaning "decide per ingress" (see
+#: :func:`_should_preserve_signature`). Named rather than spelled ``'auto'`` at each
+#: use so the sentinel has exactly one definition.
+AUTO_PRESERVE_SIGNATURE = "auto"
+
 
 def identity(x):
     """Transparent function, returning what's been input"""
@@ -205,92 +210,102 @@ def _defaults_and_kwdefaults_of_func(func: Callable):
         return sig._defaults_, sig._kwdefaults_
 
 
-def _should_preserve_signature(ingress, func, preserve_mode):
-    """Determine if signature should be auto-preserved from func to ingress.
+def _is_generic_signature(func: Callable) -> bool:
+    """Whether ``func``'s signature is exactly ``(*args, **kwargs)``.
 
-    Parameters
-    ----------
-    ingress : callable or None
-        The ingress function
-    func : callable
-        The wrapped function
-    preserve_mode : 'auto' | True | False
-        The preservation mode
+    Such a signature carries no interface information, which is what makes it safe
+    to show the wrapped function's signature in its place.
 
-    Returns
-    -------
-    bool
-        True if signature should be preserved
+    >>> _is_generic_signature(lambda *a, **kw: None)
+    True
+    >>> _is_generic_signature(lambda x, *a, **kw: None)
+    False
+    >>> _is_generic_signature(lambda *a: None)
+    False
+    >>> _is_generic_signature(lambda x: None)
+    False
     """
-    if preserve_mode is False:
+    try:
+        params = list(signature(func).parameters.values())
+    except (ValueError, TypeError):  # no inspectable signature (some builtins)
         return False
-    if preserve_mode is True:
-        return True
+    return [p.kind for p in params] == [Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD]
 
-    # 'auto' mode: preserve if ingress has generic (*args, **kwargs)
-    if ingress is None:
-        return False
 
-    try:
-        ingress_sig = signature(ingress)
-    except (ValueError, TypeError):
-        # Can't get signature, don't preserve
-        return False
+def _should_preserve_signature(ingress, preserve_mode):
+    """Whether the wrapper should advertise the wrapped function's signature instead
+    of ``ingress``'s.
+
+    :param ingress: The incoming data transformer, or ``None``
+    :param preserve_mode: ``True`` (always), ``False`` (never), or ``'auto'``
+    :return: ``True`` if the wrapped function's signature should be used
 
-    params = list(ingress_sig.parameters.values())
+    ``'auto'`` preserves only when the ingress signature is ``(*args, **kwargs)`` and
+    the ingress does not declare a ``__signature__`` of its own -- i.e. only when the
+    ingress says nothing about its interface, so there is nothing to overwrite:
 
-    # Check if ingress has exactly (*args, **kwargs) signature
-    if len(params) != 2:
+    >>> _should_preserve_signature(lambda *a, **kw: None, 'auto')
+    True
+    >>> _should_preserve_signature(lambda a: None, 'auto')
+    False
+
+    ``True``/``False`` are unconditional, and are the way to override ``'auto'``'s guess:
+
+    >>> _should_preserve_signature(lambda a: None, True)
+    True
+    >>> _should_preserve_signature(lambda *a, **kw: None, False)
+    False
+    """
+    if preserve_mode != AUTO_PRESERVE_SIGNATURE:
+        return bool(preserve_mode)
+    if ingress is None:
         return False
+    return _is_generic_signature(ingress) and not hasattr(ingress, "__signature__")
 
-    is_generic = (
-        params[0].kind == Parameter.VAR_POSITIONAL
-        and params[1].kind == Parameter.VAR_KEYWORD
-    )
 
-    # Only preserve if generic and doesn't already have __signature__
-    return is_generic and not hasattr(ingress, "__signature__")
+def _return_annotation_of(func: Callable) -> Any:
+    """The return annotation of ``func``, or ``empty`` if it has none or has no
+    inspectable signature at all."""
+    try:
+        return Sig(func).return_annotation
+    except (ValueError, TypeError):
+        return empty
 
 
 def _get_return_annotation(func, egress):
-    """Get return annotation with smart fallback logic.
+    """Resolve the return annotation a :class:`Wrap` should advertise.
 
-    Fallback chain: egress annotation → func annotation → empty
+    :param func: The wrapped function
+    :param egress: The outgoing data transformer, or ``None``
+    :return: The return annotation to use, or ``empty``
 
-    Parameters
-    ----------
-    func : callable
-        The wrapped function
-    egress : callable or None
-        The egress function
+    The egress wins when it declares a return annotation, since it is what actually
+    produces the wrapper's output. Otherwise the annotation falls back to ``func``'s,
+    on the assumption that an unannotated egress does not change the type:
 
-    Returns
-    -------
-    annotation
-        The return annotation to use, or Parameter.empty
-    """
-    func_sig = Sig(func)
-    func_return = func_sig.return_annotation
+    >>> def f(x: int) -> str: return str(x)
+    >>> def annotated_egress(out) -> bytes: return out.encode()
+    >>> def bare_egress(out): return out
 
-    if egress is None:
-        # No egress: use func's return annotation
-        return func_return if func_return is not Parameter.empty else empty
+    >>> _get_return_annotation(f, None)
+    <class 'str'>
+    >>> _get_return_annotation(f, annotated_egress)
+    <class 'bytes'>
+    >>> _get_return_annotation(f, bare_egress)
+    <class 'str'>
 
-    # Egress provided: check its annotation first
-    try:
-        egress_sig = Sig(egress)
-        egress_return = egress_sig.return_annotation
-    except (ValueError, TypeError):
-        # Can't get egress signature, fall back to func
-        return func_return if func_return is not Parameter.empty else empty
-
-    if egress_return is not Parameter.empty:
-        # Egress has annotation, use it
-        return egress_return
+    An unannotated ``func`` yields ``empty``, not ``None`` -- ``None`` is a legitimate
+    annotation meaning "returns None", so the two must stay distinguishable:
 
-    # Egress has no annotation: fall back to func's annotation
-    # Assumption: egress doesn't transform the type
-    return func_return if func_return is not Parameter.empty else empty
+    >>> def g(x): pass
+    >>> _get_return_annotation(g, None) is empty
+    True
+    """
+    if egress is not None:
+        egress_return = _return_annotation_of(egress)
+        if egress_return is not empty:
+            return egress_return
+    return _return_annotation_of(func)
 
 
 class Wrap(_Wrap):
@@ -565,7 +580,13 @@ class Wrap(_Wrap):
     """
 
     def __init__(
-        self, func, ingress=None, egress=None, *, name=None, preserve_signature="auto"
+        self,
+        func,
+        ingress=None,
+        egress=None,
+        *,
+        name=None,
+        preserve_signature=AUTO_PRESERVE_SIGNATURE,
     ):
         super().__init__(func, ingress, egress, name=name)
         ingress_sig = Sig(func)
@@ -576,7 +597,6 @@ def __init__(
                 func
             )
         else:
-
             if isinstance(ingress, MakeFromFunc):
                 func_to_ingress = ingress  # it's not the ingress function itself
                 # ... but an ingress factory: Should make the ingress in function of func
@@ -585,15 +605,19 @@ def __init__(
                 assert callable(ingress), f"Should be callable: {ingress}"
                 self.ingress = ingress
 
-            # Apply signature preservation if needed
-            if _should_preserve_signature(self.ingress, func, preserve_signature):
-                # Preserve signature and annotations from func to ingress
-                self.ingress.__signature__ = signature(func)
-                self.ingress.__annotations__ = getattr(func, "__annotations__", {})
-
-            ingress_sig = Sig(self.ingress)
+            # Preserving means *this wrapper* presents func's interface. Read it from
+            # func directly; never write it onto self.ingress. The ingress belongs to
+            # the caller -- a decorator typically defines one and reuses it for every
+            # function it wraps -- so stamping __signature__ on it corrupted both the
+            # caller's object and every later Wrap built from the same ingress.
+            signature_source = (
+                func
+                if _should_preserve_signature(self.ingress, preserve_signature)
+                else self.ingress
+            )
+            ingress_sig = Sig(signature_source)
             self.__defaults__, self.__kwdefaults__ = _defaults_and_kwdefaults_of_func(
-                self.ingress
+                signature_source
             )
 
         # Set egress
@@ -974,9 +998,9 @@ def name_map(cls, wrapped, **old_to_new_name):
 
         """
         new_to_old_name = {v: k for k, v in old_to_new_name.items()}
-        assert len(new_to_old_name) == len(
-            old_to_new_name
-        ), f"Inversion is not possible since {old_to_new_name=} has duplicate values."
+        assert len(new_to_old_name) == len(old_to_new_name), (
+            f"Inversion is not possible since {old_to_new_name=} has duplicate values."
+        )
         return cls(
             wrapped,
             partial(Pipe(items_with_mapped_keys, dict), key_mapper=new_to_old_name),
@@ -2541,9 +2565,9 @@ def add_smart_defaults(
 
     """
     names_not_in_func_arguments = smart_defaults.keys() - Sig(func).names
-    assert (
-        not names_not_in_func_arguments
-    ), f"These weren't argument names of the {func} function: {names_not_in_func_arguments}"
+    assert not names_not_in_func_arguments, (
+        f"These weren't argument names of the {func} function: {names_not_in_func_arguments}"
+    )
     kwargs_trans = partial(
         complete_dict_applying_functions,
         _only_if_name_missing=_only_if_name_missing,