fix(resolver): preserve qualified call resolution

src/codegraph/builder.py

@@ -16,7 +16,7 @@
 from .scanner.walk import discover_files
 from .store import IndexStore
 
-_RESOLVER_VERSION = 2
+_RESOLVER_VERSION = 5
 
 
 @dataclass

src/codegraph/resolver.py

@@ -61,16 +61,82 @@ def last_segment(name: str) -> str:
     return name
 
 
+def _unique_local_symbol_id(store: IndexStore, file_id: int, qualname: str):
+    """Return a local symbol id only when its qualified name is unique."""
+    rows = store.conn.execute(
+        "SELECT id FROM symbols WHERE file_id = ? AND qualname = ?",
+        (file_id, qualname),
+    ).fetchall()
+    if len(rows) == 1:
+        return rows[0]["id"]
+    return None
+
+
+def _local_qualified_symbol(store: IndexStore, file_id: int,
+                            qualified: str, name: str, caller_name=None):
+    """Resolve a qualified target in the caller's lexical scopes."""
+    file = store.file_by_id(file_id)
+    if file is None:
+        return None
+    module = file["module"]
+    caller = None
+    if caller_name:
+        caller = store.conn.execute(
+            "SELECT kind, parent, qualname FROM symbols "
+            "WHERE file_id = ? AND qualname = ? ORDER BY id LIMIT 1",
+            (file_id, caller_name),
+        ).fetchone()
+
+    if caller is not None:
+        scope = caller["qualname"] if caller["kind"] == "class" \
+            else caller["parent"]
+    elif caller_name and "." in caller_name:
+        scope = caller_name.rsplit(".", 1)[0]
+    else:
+        scope = module
+    if not scope:
+        scope = module
+
+    receiver = qualified.split(".", 1)[0]
+    if receiver in ("self", "this", "Self"):
+        if caller is None or not scope:
+            return None
+        owner = store.conn.execute(
+            "SELECT kind FROM symbols WHERE file_id = ? AND qualname = ? "
+            "ORDER BY id LIMIT 1",
+            (file_id, scope),
+        ).fetchone()
+        if owner is None or owner["kind"] not in (
+                "class", "interface", "type"):
+            return None
+        return _unique_local_symbol_id(store, file_id, f"{scope}.{name}")
+
+    local_id = _unique_local_symbol_id(store, file_id, qualified)
+    if local_id is not None:
+        return local_id
+
+    while scope:
+        local_id = _unique_local_symbol_id(
+            store, file_id, f"{scope}.{qualified}")
+        if local_id is not None:
+            return local_id
+        if scope == module or "." not in scope:
+            break
+        scope = scope.rsplit(".", 1)[0]
+    return None
+
+
 def _root_of(store: IndexStore) -> Path:
     return Path(store.get_meta("root") or ".")
 
 
 def resolve_callee(store: IndexStore, file_id: int, callee_text: str,
-                   blocked_file_ids=()):
+                   blocked_file_ids=(), caller_name=None):
     """Return the symbol id a call target refers to, or None.
 
     ``blocked_file_ids`` prevents fallback to symbols in import targets that
     are not the selected candidate for the import.
+    ``caller_name`` scopes local qualified-owner resolution to the call site.
     """
     name = last_segment(callee_text)
     if not name:
@@ -80,18 +146,35 @@ def resolve_callee(store: IndexStore, file_id: int, callee_text: str,
         return None
     blocked_file_ids = set(blocked_file_ids)
 
-    # 1. same file: exact qualname, then unique name
-    row = store.conn.execute(
-        "SELECT id FROM symbols WHERE file_id = ? AND qualname = ? ORDER BY id LIMIT 1",
-        (file_id, callee_text),
-    ).fetchone()
-    if row:
-        return row["id"]
+    # 1. same file: exact qualname, local qualified owner, then unique bare name
     rows = store.conn.execute(
-        "SELECT id FROM symbols WHERE file_id = ? AND name = ?", (file_id, name)
+        "SELECT id FROM symbols WHERE file_id = ? AND qualname = ?",
+        (file_id, callee_text),
     ).fetchall()
     if len(rows) == 1:
         return rows[0]["id"]
+    if callee_text == name:
+        rows = store.conn.execute(
+            "SELECT id FROM symbols WHERE file_id = ? AND name = ?",
+            (file_id, name),
+        ).fetchall()
+        if len(rows) == 1:
+            return rows[0]["id"]
+    else:
+        qualified = callee_text.replace("::", ".")
+        local_id = _local_qualified_symbol(
+            store, file_id, qualified, name, caller_name)
+        if local_id is not None:
+            return local_id
+        receiver = qualified.split(".", 1)[0]
+        if receiver in ("self", "this", "Self"):
+            rows = store.conn.execute(
+                "SELECT id FROM symbols WHERE file_id = ? AND name = ? "
+                "AND kind = 'method'",
+                (file_id, name),
+            ).fetchall()
+            if len(rows) == 1:
+                return rows[0]["id"]
 
     alias_target = _rust_alias_symbol(store, file_id, callee_text)
     if alias_target is not None:
@@ -1168,6 +1251,7 @@ def resolve_all(store: IndexStore, file_ids=None, call_ids=(), import_ids=(),
                 row["file_id"],
                 row["callee"],
                 blocked_file_ids=blocked_import_files.get(row["file_id"], ()),
+                caller_name=row["caller_name"],
             )
             store.conn.execute(
                 "UPDATE calls SET caller_id = ?, callee_id = ? WHERE id = ?",

tests/test_builder.py

@@ -170,6 +170,160 @@ def test_resolver_version_change_re_resolves_fallback_import(self):
             finally:
                 store.close()
 
+    def test_resolver_version_bump_rechecks_stale_qualified_call(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "lib.py").write_text(
+                "def f():\n    return 'imported'\n", encoding="utf-8")
+            (root / "app.py").write_text(
+                "import lib\n\n"
+                "def f():\n    return 'local'\n\n"
+                "def caller():\n    return lib.f()\n",
+                encoding="utf-8",
+            )
+
+            cfg = load_config(root=str(root))
+            cfg.engine = "quick"
+            build_index(cfg)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                file_id = store.file_by_path("app.py")["id"]
+                call = store.find_call(callee="lib.f", file_id=file_id)
+                self.assertIsNotNone(call)
+                local_id = store.symbol_by_qualname("app.f")["id"]
+                self.assertEqual(
+                    store.symbol_by_id(call["callee_id"])["qualname"], "lib.f"
+                )
+                store.conn.execute(
+                    "UPDATE calls SET callee_id = ? WHERE id = ?",
+                    (local_id, call["id"]),
+                )
+                scan_config = json.loads(store.get_meta("scan_config"))
+                # Version 2 indexes predate the qualified-call resolution fix.
+                scan_config["resolver_version"] = 2
+                store.set_meta("scan_config", json.dumps(scan_config))
+                store.conn.commit()
+            finally:
+                store.close()
+
+            report = build_index(cfg)
+            self.assertEqual(report.files_changed, 0)
+            self.assertEqual(report.files_skipped, 2)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                call = store.find_call(callee="lib.f", file_id=file_id)
+                self.assertEqual(
+                    store.symbol_by_id(call["callee_id"])["qualname"], "lib.f"
+                )
+            finally:
+                store.close()
+
+    def test_resolver_version_bump_rechecks_nested_qualified_call(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "lib.py").write_text(
+                "class C:\n"
+                "    def helper(self):\n"
+                "        return 'imported'\n",
+                encoding="utf-8",
+            )
+            (root / "app.py").write_text(
+                "from lib import C\n\n"
+                "class Outer:\n"
+                "    class C:\n"
+                "        def helper(self):\n"
+                "            return 'nested'\n\n"
+                "def caller():\n"
+                "    return C.helper(None)\n",
+                encoding="utf-8",
+            )
+
+            cfg = load_config(root=str(root))
+            cfg.engine = "quick"
+            build_index(cfg)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                file_id = store.file_by_path("app.py")["id"]
+                call = store.find_call(callee="C.helper", file_id=file_id)
+                nested_id = store.symbol_by_qualname("app.Outer.C.helper")["id"]
+                store.conn.execute(
+                    "UPDATE calls SET callee_id = ? WHERE id = ?",
+                    (nested_id, call["id"]),
+                )
+                scan_config = json.loads(store.get_meta("scan_config"))
+                # Version 3 indexes can contain the old unrestricted suffix match.
+                scan_config["resolver_version"] = 3
+                store.set_meta("scan_config", json.dumps(scan_config))
+                store.conn.commit()
+            finally:
+                store.close()
+
+            report = build_index(cfg)
+            self.assertEqual(report.files_changed, 0)
+            self.assertEqual(report.files_skipped, 2)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                call = store.find_call(callee="C.helper", file_id=file_id)
+                self.assertEqual(
+                    store.symbol_by_id(call["callee_id"])["qualname"],
+                    "lib.C.helper",
+                )
+            finally:
+                store.close()
+
+    def test_resolver_version_bump_rechecks_ambiguous_qualified_call(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / "Caller.java").write_text(
+                "package app;\n\n"
+                "class C {\n"
+                "    static void f() {}\n"
+                "    static void f(int value) {}\n"
+                "}\n\n"
+                "class Caller {\n"
+                "    void invoke() {\n"
+                "        C.f();\n"
+                "    }\n"
+                "}\n",
+                encoding="utf-8",
+            )
+
+            cfg = load_config(root=str(root))
+            cfg.engine = "quick"
+            build_index(cfg)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                file_id = store.file_by_path("Caller.java")["id"]
+                call = store.find_call(callee="C.f", file_id=file_id)
+                self.assertIsNotNone(call)
+                stale_id = store.symbol_by_qualname("app.C.f")["id"]
+                store.conn.execute(
+                    "UPDATE calls SET callee_id = ? WHERE id = ?",
+                    (stale_id, call["id"]),
+                )
+                scan_config = json.loads(store.get_meta("scan_config"))
+                scan_config["resolver_version"] = 4
+                store.set_meta("scan_config", json.dumps(scan_config))
+                store.conn.commit()
+            finally:
+                store.close()
+
+            report = build_index(cfg)
+            self.assertEqual(report.files_changed, 0)
+            self.assertEqual(report.files_skipped, 1)
+
+            store = IndexStore(str(cfg.db_path))
+            try:
+                call = store.find_call(callee="C.f", file_id=file_id)
+                self.assertIsNone(call["callee_id"])
+            finally:
+                store.close()
+
     def test_changed_file_reparsed_only(self):
         build_index(self._cfg())
         target = self.root / "pkg" / "pricing.py"