fix(structural): bind comprehension targets

diffgraph/structural.py

@@ -92,6 +92,7 @@ class _Call:
     name: str
     line: int
     snippet: str
+    comprehension_bindings: Tuple[str, ...] = ()
 
 
 def _blob(repository: str, oid: Optional[str]) -> Optional[bytes]:
@@ -325,6 +326,55 @@ def case_pattern_identifiers(node) -> set:
             found.update(case_pattern_identifiers(child))
         return found
 
+    def enclosing_comprehension_bindings(node) -> Tuple[str, ...]:
+        """Return comprehension targets bound before ``node`` executes.
+
+        A comprehension target is not visible in its own iterable or in an
+        earlier clause's iterable. Tree-sitter stores each ``for_in_clause``
+        beside the expression it governs, so derive visibility from the call's
+        position within the direct clauses rather than treating every target as
+        an enclosing-function binding.
+        """
+        found = set()
+        ancestor = node.parent
+        while ancestor is not None:
+            if ancestor.type in (
+                "list_comprehension",
+                "set_comprehension",
+                "dictionary_comprehension",
+                "generator_expression",
+            ):
+                clauses = [
+                    child
+                    for child in ancestor.children
+                    if child.type in ("for_in_clause", "if_clause")
+                ]
+                visible_clauses = clauses
+                for index, clause in enumerate(clauses):
+                    if not (
+                        clause.start_byte <= node.start_byte
+                        and node.end_byte <= clause.end_byte
+                    ):
+                        continue
+                    iterable = clause.child_by_field_name("right")
+                    if (
+                        iterable is not None
+                        and iterable.start_byte <= node.start_byte
+                        and node.end_byte <= iterable.end_byte
+                    ):
+                        visible_clauses = clauses[:index]
+                    else:
+                        visible_clauses = clauses[: index + 1]
+                    break
+                for clause in visible_clauses:
+                    if clause.type != "for_in_clause":
+                        continue
+                    left = clause.child_by_field_name("left")
+                    if left is not None:
+                        found.update(identifiers(left))
+            ancestor = ancestor.parent
+        return tuple(sorted(found))
+
     def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
         next_parents = parents
         if node.type in ("class_definition", "function_definition"):
@@ -498,6 +548,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
                         _node_text(content, function),
                         node.start_point[0] + 1,
                         _node_text(content, node),
+                        enclosing_comprehension_bindings(node),
                     )
                 )
 
@@ -646,6 +697,8 @@ def _resolve_call_target(
     are deliberately left unresolved. Attribute calls and ambiguous duplicate
     definitions are likewise omitted rather than guessed.
     """
+    if call.name in call.comprehension_bindings:
+        return None
     candidates: List[str] = []
     current_name = call.caller
     while current_name is not None:

tests/test_structural.py

@@ -1492,6 +1492,75 @@ def test_as_pattern_bindings_do_not_create_import_grounded_call_edges(tmp_path):
     assert calls == []
 
 
+def test_comprehension_targets_shadow_imports_only_inside_comprehensions(tmp_path):
+    """Comprehension targets shadow imports without leaking into their function."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "comprehension_bindings.py",
+        "from remote.worker import execute as run_remote\n\n"
+        "def build(values):\n"
+        "    run_remote()\n"
+        "    result = [run_remote() for run_remote in values]\n"
+        "    run_remote()\n"
+        "    return result\n",
+    )
+    git(root, "add", "comprehension_bindings.py")
+
+    artifact = analyze_local_diff(str(root), staged=True)
+
+    assert_valid(artifact)
+    calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
+    assert len(calls) == 2
+    assert {item["evidence"][0]["line_start"] for item in calls} == {4, 6}
+    assert all(item["resolution_method"] == "import_grounded" for item in calls)
+
+
+def test_comprehension_clauses_bind_targets_in_evaluation_order(tmp_path):
+    """Comprehension iterables see only targets from earlier clauses."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "comprehension_clause_order.py",
+        "from remote.worker import execute as run_remote\n\n"
+        "def build(values):\n"
+        "    own_iterable = [item for run_remote in run_remote()]\n"
+        "    earlier_iterable = [item for item in run_remote() for run_remote in values]\n"
+        "    later_iterable = [item for run_remote in values for item in run_remote()]\n"
+        "    return own_iterable, earlier_iterable, later_iterable\n",
+    )
+    git(root, "add", "comprehension_clause_order.py")
+
+    artifact = analyze_local_diff(str(root), staged=True)
+
+    assert_valid(artifact)
+    calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
+    assert {item["evidence"][0]["line_start"] for item in calls} == {4, 5}
+    assert all(item["resolution_method"] == "import_grounded" for item in calls)
+
+
+def test_comprehension_filter_does_not_bind_later_targets(tmp_path):
+    """A filter sees prior targets but not names bound by later clauses."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "comprehension_filter_order.py",
+        "from remote.worker import execute as run_remote\n\n"
+        "def build(values, sources):\n"
+        "    return [item for item in values "
+        "if run_remote() for run_remote in sources]\n",
+    )
+    git(root, "add", "comprehension_filter_order.py")
+
+    artifact = analyze_local_diff(str(root), staged=True)
+
+    assert_valid(artifact)
+    calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
+    assert len(calls) == 1
+    assert calls[0]["evidence"][0]["line_start"] == 4
+    assert calls[0]["resolution_method"] == "import_grounded"
+
+
 def test_match_pattern_captures_do_not_create_import_grounded_call_edges(tmp_path):
     """Python match captures, including splats, shadow imports in case bodies."""
     root = repo(tmp_path)