fix: require pure Python forwarding shape

desloppify/languages/python/extractors.py

@@ -1,5 +1,6 @@
 """Python extraction: function bodies, class structure, param patterns."""
 
+import ast
 import hashlib
 import re
 from pathlib import Path
@@ -125,42 +126,131 @@ def py_passthrough_pattern(name: str) -> str:
     return rf"\b{escaped}\s*=\s*{escaped}\b"
 
 
-_PY_DEF_RE = re.compile(r"^def\s+(\w+)\s*\(", re.MULTILINE)
+def _function_param_names(node: ast.FunctionDef) -> list[str]:
+    """Return the explicit parameter names that count toward wrapper forwarding."""
+
+    arguments = [
+        *node.args.posonlyargs,
+        *node.args.args,
+        *node.args.kwonlyargs,
+    ]
+    if node.args.vararg is not None:
+        arguments.append(node.args.vararg)
+    if node.args.kwarg is not None:
+        arguments.append(node.args.kwarg)
+    return [
+        argument.arg for argument in arguments if argument.arg not in {"self", "cls"}
+    ]
+
+
+def _is_docstring_expr(statement: ast.stmt) -> bool:
+    """Return whether a statement is a function docstring expression."""
+
+    return (
+        isinstance(statement, ast.Expr)
+        and isinstance(statement.value, ast.Constant)
+        and isinstance(statement.value.value, str)
+    )
+
+
+def _is_direct_parameter_value(value: ast.expr, parameter_names: set[str]) -> bool:
+    """Return whether a call argument directly forwards one function parameter."""
+
+    if isinstance(value, ast.Name):
+        return value.id in parameter_names
+    return (
+        isinstance(value, ast.Starred)
+        and isinstance(value.value, ast.Name)
+        and value.value.id in parameter_names
+    )
+
+
+def _is_direct_call_target(target: ast.expr) -> bool:
+    """Return whether a call target is a direct non-constructor name chain.
+
+    A forwarding call to a lower-case function can be redundant, while a
+    Capitalized target conventionally constructs a value.  Constructor
+    factories are meaningful transformations, even when every field is
+    supplied from a same-named parameter.
+    """
+
+    names: list[str] = []
+    while isinstance(target, ast.Attribute):
+        names.append(target.attr)
+        target = target.value
+    if not isinstance(target, ast.Name):
+        return False
+    names.append(target.id)
+    return not any(name[:1].isupper() for name in names)
+
+
+def _passthrough_return_statement(
+    node: ast.FunctionDef,
+    parameter_names: list[str],
+) -> ast.Return | None:
+    """Return the sole direct-forwarding return statement for a pure wrapper.
+
+    Decorated entrypoints and multi-step functions can forward many arguments
+    while still constructing requests, making decisions, or retaining state.
+    They are not the redundant wrappers this detector is meant to surface.
+    """
+
+    if node.decorator_list:
+        return None
+    statements = list(node.body)
+    if statements and _is_docstring_expr(statements[0]):
+        statements = statements[1:]
+    if len(statements) != 1 or not isinstance(statements[0], ast.Return):
+        return None
+    statement = statements[0]
+    if not isinstance(statement.value, ast.Call):
+        return None
+    if not _is_direct_call_target(statement.value.func):
+        return None
+
+    values = [
+        *statement.value.args,
+        *(keyword.value for keyword in statement.value.keywords),
+    ]
+    if not values:
+        return None
+    parameter_name_set = set(parameter_names)
+    if not all(
+        _is_direct_parameter_value(value, parameter_name_set) for value in values
+    ):
+        return None
+    return statement
 
 
 def detect_passthrough_functions(path: Path) -> list[dict]:
-    """Detect Python functions where most params are same-name forwarded."""
+    """Detect pure Python wrappers where most params are same-name forwarded."""
     entries = []
     for filepath in find_py_files(path):
         content = read_file(filepath)
         if content is None:
             continue
-        for m in _PY_DEF_RE.finditer(content):
-            name = m.group(1)
-            depth = 1
-            i = m.end()
-            while i < len(content) and depth > 0:
-                ch = content[i]
-                if ch == "(":
-                    depth += 1
-                elif ch == ")":
-                    depth -= 1
-                i += 1
-            if depth != 0:
+        try:
+            tree = ast.parse(content, filename=filepath)
+        except SyntaxError:
+            continue
+        for node in tree.body:
+            if not isinstance(node, ast.FunctionDef):
                 continue
-            param_str = content[m.end() : i - 1]
-            params = extract_py_params(param_str)
+            name = node.name
+            params = _function_param_names(node)
             if len(params) < 4:
                 continue
-            rest_after_paren = content[i:]
-            colon_m = re.search(r":", rest_after_paren)
-            if not colon_m:
+            return_statement = _passthrough_return_statement(node, params)
+            if return_statement is None:
+                continue
+            body = ast.get_source_segment(content, return_statement)
+            if body is None:
+                continue
+            call = return_statement.value
+            if not isinstance(call, ast.Call):
                 continue
-            rest = rest_after_paren[colon_m.end() :]
-            bm = re.search(r"\n(?=[^\s\n#])", rest)
-            body = rest[: bm.start()] if bm else rest
 
-            has_kwargs_spread = bool(re.search(r"\*\*kwargs\b", body))
+            has_kwargs_spread = any(keyword.arg is None for keyword in call.keywords)
             pt, direct = classify_params(
                 params, body, py_passthrough_pattern, occurrences_per_match=2
             )
@@ -183,7 +273,7 @@ def detect_passthrough_functions(path: Path) -> list[dict]:
                     "passthrough": len(pt),
                     "direct": len(direct),
                     "ratio": round(ratio, 2),
-                    "line": content[: m.start()].count("\n") + 1,
+                    "line": node.lineno,
                     "tier": tier,
                     "confidence": confidence,
                     "passthrough_params": sorted(pt),

desloppify/languages/python/tests/test_py_extractors.py

@@ -350,7 +350,7 @@ def test_base_classes_extracted(self, tmp_path):
 
 
 class TestDetectPassthrough:
-    def test_passthrough_detected(self, tmp_path):
+    def test_lowercase_function_wrapper_detected(self, tmp_path):
         fp = tmp_path / "pt.py"
         fp.write_text(
             textwrap.dedent("""\
@@ -359,9 +359,186 @@ def wrapper(a, b, c, d, e):
         """)
         )
         entries = detect_passthrough_functions(tmp_path)
-        if entries:
-            assert entries[0]["function"] == "wrapper"
-            assert entries[0]["passthrough"] >= 4
+        wrapper = next(entry for entry in entries if entry["function"] == "wrapper")
+        assert wrapper["passthrough"] >= 4
+
+    def test_documented_passthrough_detected(self, tmp_path):
+        fp = tmp_path / "documented.py"
+        fp.write_text(
+            textwrap.dedent('''\
+            def wrapper(a, b, c, d, e):
+                """Backward-compatible forwarding alias."""
+                return inner(a=a, b=b, c=c, d=d, e=e)
+        ''')
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "wrapper" in names
+
+    def test_decorated_entrypoint_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "entrypoint.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            @register_command
+            @with_application_context
+            def seed(a, b, c, d, e):
+                request = SeedRequest(a=a, b=b, c=c, d=d, e=e)
+                run_seed(request)
+        """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "seed" not in names
+
+    def test_decorated_pure_wrapper_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "decorated_wrapper.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            @register_command
+            def callback(a, b, c, d, e):
+                return handle(a=a, b=b, c=c, d=d, e=e)
+        """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "callback" not in names
+
+    def test_multi_step_decision_router_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "decision.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            def decide(model, *, receiver_ip3, threshold, evidence, profile_required, unmodeled_orders):
+                power_informed = power_decision(
+                    model,
+                    threshold=threshold,
+                    evidence=evidence,
+                    profile_required=profile_required,
+                )
+                if power_informed is not None:
+                    return power_informed
+                return fallback_decision(
+                    model,
+                    receiver_ip3=receiver_ip3,
+                    threshold=threshold,
+                    evidence=evidence,
+                    profile_required=profile_required,
+                    unmodeled_orders=unmodeled_orders,
+                )
+        """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "decide" not in names
+
+    def test_looping_validator_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "validator.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            def validate_regions(shaped, *, errors, point_to_xy_fn, polygon_points_fn, point_in_polygon_fn):
+                region_keys = set()
+                for region in shaped.regions:
+                    polygon = polygon_points_fn(region)
+                    point = point_to_xy_fn(region.get("representative_point"))
+                    if point is not None and not point_in_polygon_fn(point, polygon):
+                        errors.append("representative point outside polygon")
+                    validate_region(
+                        region,
+                        errors=errors,
+                        point_to_xy_fn=point_to_xy_fn,
+                        polygon_points_fn=polygon_points_fn,
+                        point_in_polygon_fn=point_in_polygon_fn,
+                    )
+                    region_keys.add(region["region_key"])
+                return region_keys
+        """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "validate_regions" not in names
+
+    def test_request_construction_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "request_builder.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            def build_request(a, b, c, d, e):
+                return submit(Request(a=a, b=b, c=c, d=d, e=e))
+        """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "build_request" not in names
+
+    def test_typed_dataclass_fixture_factory_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "runtime_hooks.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            from dataclasses import dataclass
+
+
+            @dataclass(frozen=True)
+            class AutoCoordinationRunRuntimeHooks:
+                command_bus: object
+                coordinator: object
+                event_store: object
+                logger: object
+                metrics: object
+                run_id: object
+
+
+            def make_runtime_hooks(
+                command_bus: object,
+                coordinator: object,
+                event_store: object,
+                logger: object,
+                metrics: object,
+                run_id: object,
+            ) -> AutoCoordinationRunRuntimeHooks:
+                return AutoCoordinationRunRuntimeHooks(
+                    command_bus=command_bus,
+                    coordinator=coordinator,
+                    event_store=event_store,
+                    logger=logger,
+                    metrics=metrics,
+                    run_id=run_id,
+                )
+            """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "make_runtime_hooks" not in names
+
+    def test_constructed_call_target_is_not_flagged(self, tmp_path):
+        fp = tmp_path / "request_submitter.py"
+        fp.write_text(
+            textwrap.dedent("""\
+            def submit_via_request(a, b, c, d, e):
+                return Request(a=a, b=b, c=c, d=d, e=e).submit(
+                    a=a, b=b, c=c, d=d, e=e,
+                )
+            """)
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert "submit_via_request" not in names
+
+    def test_syntax_error_file_is_skipped(self, tmp_path):
+        (tmp_path / "broken.py").write_text("def broken(\n")
+        (tmp_path / "wrapper.py").write_text(
+            "def wrapper(a, b, c, d, e):\n"
+            "    return inner(a=a, b=b, c=c, d=d, e=e)\n"
+        )
+
+        names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)]
+
+        assert names == ["wrapper"]
 
     def test_non_passthrough_not_flagged(self, tmp_path):
         fp = tmp_path / "real.py"