fix(structural): track local import rebindings

diffgraph/structural.py

@@ -260,7 +260,7 @@ def _parse_python(
     imports: List[_Import] = []
     calls: List[_Call] = []
     bindings: Dict[Optional[str], set] = {}
-    module_rebindings: List[Tuple[str, int]] = []
+    rebindings: List[Tuple[Optional[str], str, int]] = []
     symbol_occurrences: Dict[str, int] = {}
 
     def identifiers(node) -> set:
@@ -282,6 +282,17 @@ def as_target_identifiers(node) -> set:
             found.update(as_target_identifiers(child))
         return found
 
+    def assignment_target_identifiers(node) -> set:
+        """Return names assigned by an assignment target, not names it reads."""
+        if node.type in ("attribute", "subscript"):
+            return set()
+        if node.type == "identifier":
+            return {_node_text(content, node)}
+        found = set()
+        for child in node.children:
+            found.update(assignment_target_identifiers(child))
+        return found
+
     def case_pattern_identifiers(node) -> set:
         """Return capture names assigned by a Python structural match pattern.
 
@@ -399,6 +410,12 @@ def is_declaration_header(node) -> bool:
             )
         return False
 
+    def record_rebindings(
+        scope: Optional[str], names: set, position: int
+    ) -> None:
+        """Record lexical rebindings after the expression that creates them."""
+        rebindings.extend((scope, name, position) for name in names)
+
     def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
         next_parents = parents
         if node.type in ("class_definition", "function_definition"):
@@ -447,17 +464,15 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
                     )
                 )
                 next_parents = (*parents, (qname, kind))
-                if not parents:
-                    # A top-level declaration overwrites an imported binding at
-                    # runtime just like a top-level assignment does, but only
-                    # after the declaration header has been evaluated.
-                    body = node.child_by_field_name("body")
-                    binding_position = (
-                        body.start_byte
-                        if node.type == "function_definition" and body is not None
-                        else node.end_byte
-                    )
-                    module_rebindings.append((name, binding_position))
+                # A declaration overwrites an imported binding in its enclosing
+                # scope, but only after its declaration header is evaluated.
+                body = node.child_by_field_name("body")
+                binding_position = (
+                    body.start_byte
+                    if node.type == "function_definition" and body is not None
+                    else node.end_byte
+                )
+                record_rebindings(parent, {name}, binding_position)
         elif node.type in ("import_statement", "import_from_statement"):
             snippet = _node_text(content, node)
             if node.type == "import_statement":
@@ -594,15 +609,13 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
             if name_node is not None and name_node.type == "identifier":
                 name = _node_text(content, name_node)
                 bindings.setdefault(scope, set()).add(name)
-                if scope is None:
-                    module_rebindings.append((name, node.start_byte))
+                record_rebindings(scope, {name}, node.start_byte)
         elif node.type == "case_clause":
             # Capture patterns bind their names before the case body executes.
             # Keep them lexical and line-aware just like assignment bindings.
             bound_names = case_pattern_identifiers(node)
             bindings.setdefault(scope, set()).update(bound_names)
-            if scope is None:
-                module_rebindings.extend((name, node.start_byte) for name in bound_names)
+            record_rebindings(scope, bound_names, node.start_byte)
         elif node.type in (
             "assignment", "annotated_assignment", "for_statement", "named_expression"
         ):
@@ -614,7 +627,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
                 "name" if node.type == "named_expression" else "left"
             )
             if left is not None:
-                bound_names = identifiers(left)
+                bound_names = assignment_target_identifiers(left)
                 binding_scope = scope
                 if node.type == "named_expression":
                     # Lambdas have their own lexical scope, but are not
@@ -625,15 +638,16 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
                     elif is_declaration_header(node):
                         binding_scope = parents[-2][0] if len(parents) > 1 else None
                 bindings.setdefault(binding_scope, set()).update(bound_names)
-                if binding_scope is None:
-                    binding_position = node.end_byte
-                    if node.type == "for_statement":
-                        iterable = node.child_by_field_name("right")
-                        if iterable is not None:
-                            binding_position = iterable.end_byte
-                    module_rebindings.extend(
-                        (name, binding_position) for name in bound_names
-                    )
+                binding_position = node.end_byte
+                if node.type == "for_statement":
+                    iterable = node.child_by_field_name("right")
+                    if iterable is not None:
+                        binding_position = iterable.end_byte
+                annotation_only = node.type in (
+                    "assignment", "annotated_assignment"
+                ) and not any(child.type == "=" for child in node.children)
+                if not annotation_only:
+                    record_rebindings(binding_scope, bound_names, binding_position)
         elif (
             node.type == "as_pattern"
             and node.parent is not None
@@ -650,8 +664,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
             if target is not None:
                 bound_names = as_target_identifiers(target)
                 bindings.setdefault(scope, set()).update(bound_names)
-                if scope is None:
-                    module_rebindings.extend((name, node.start_byte) for name in bound_names)
+                record_rebindings(scope, bound_names, target.start_byte)
         for child in node.children:
             visit(child, next_parents)
 
@@ -661,7 +674,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
         sorted(imports, key=lambda item: (item.line, item.module, item.snippet)),
         sorted(calls, key=lambda item: (item.line, item.caller or "", item.name, item.snippet)),
         bindings,
-        module_rebindings,
+        rebindings,
     )
 
 
@@ -801,7 +814,7 @@ def _resolve_call_target(
 
 def _imported_call_targets(
     imports: Dict[Tuple[str, int], _Import],
-    module_rebindings: List[Tuple[str, int]],
+    rebindings: List[Tuple[Optional[str], str, int]],
 ) -> Dict[Optional[str], Dict[str, List[Tuple[int, Optional[str]]]]]:
     """Map import bindings to conservative, lexical and line-aware histories.
 
@@ -824,10 +837,10 @@ def _imported_call_targets(
                 item.position, None if binding in scope_bindings else target
             ))
             scope_bindings.add(binding)
-    for binding, position in module_rebindings:
+    for scope, binding, position in rebindings:
         # A declaration, assignment, or loop target replaces the imported
         # binding after its source expression has been evaluated.
-        targets.setdefault(None, {}).setdefault(binding, []).append((position, None))
+        targets.setdefault(scope, {}).setdefault(binding, []).append((position, None))
     for scope_targets in targets.values():
         for history in scope_targets.values():
             history.sort(key=lambda item: item[0])

tests/test_structural.py

@@ -1492,6 +1492,93 @@ def test_as_pattern_bindings_do_not_create_import_grounded_call_edges(tmp_path):
     assert calls == []
 
 
+def test_as_binding_keeps_import_visible_in_context_expression(tmp_path):
+    """An ``as`` alias binds only after its context expression runs."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "as_binding_order.py",
+        "from remote.worker import execute as run_remote\n\n"
+        "with run_remote() as run_remote:\n"
+        "    run_remote()\n",
+    )
+    git(root, "add", "as_binding_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]["resolution_method"] == "import_grounded"
+    assert calls[0]["evidence"][0]["line_start"] == 3
+
+
+def test_local_import_rebinding_stops_import_grounded_call_edges(tmp_path):
+    """A local assignment must replace a same-scope imported call target."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "local_import_rebinding.py",
+        "def caller():\n"
+        "    from remote.worker import execute as run_remote\n"
+        "    run_remote()\n"
+        "    run_remote = lambda: None\n"
+        "    run_remote()\n",
+    )
+    git(root, "add", "local_import_rebinding.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]["source_id"] == "sym::local_import_rebinding.py::caller"
+    assert calls[0]["resolution_method"] == "import_grounded"
+    assert calls[0]["evidence"][0]["line_start"] == 3
+
+
+def test_attribute_and_annotation_only_assignments_do_not_shadow_imports(tmp_path):
+    """Only assignment targets with a value replace an imported binding."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "assignment_binding_precision.py",
+        "from remote.worker import execute as run_remote\n\n"
+        "run_remote.result = None\n"
+        "run_remote: object\n"
+        "run_remote()\n",
+    )
+    git(root, "add", "assignment_binding_precision.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]["resolution_method"] == "import_grounded"
+
+
+@pytest.mark.parametrize("declaration", ["def run_remote():\n    pass", "class run_remote:\n    pass"])
+def test_nested_declaration_rebinds_function_local_import(tmp_path, declaration):
+    """Nested declarations replace same-scope imported call targets."""
+    root = repo(tmp_path)
+    write(
+        root,
+        "nested_declaration_rebind.py",
+        "def caller():\n"
+        "    from remote.worker import execute as run_remote\n\n"
+        + "    " + declaration.replace("\n", "\n    ") + "\n\n"
+        "    run_remote()\n",
+    )
+    git(root, "add", "nested_declaration_rebind.py")
+
+    artifact = analyze_local_diff(str(root), staged=True)
+
+    assert_valid(artifact)
+    calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
+    assert calls == []
+
+
 def test_named_expression_bindings_do_not_create_import_grounded_call_edges(tmp_path):
     """A walrus target shadows an import for calls after the assignment."""
     root = repo(tmp_path)