diff --git a/graphify/extract.py b/graphify/extract.py index 2ff5411cf..42c768637 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -2678,6 +2678,651 @@ def key(label: str) -> str: }) +def _resolve_kotlin_member_calls( + per_file: list[dict], + all_nodes: list[dict], + all_edges: list[dict], +) -> None: + """Resolve Kotlin member calls through method-scoped receiver type facts. + + Kotlin member calls are withheld from bare-name matching because a common + method name can belong to many unrelated classes. Resolution succeeds only + for one real JVM type definition and one method owned by that type. + """ + + def key(label: str) -> str: + return str(label).strip().removeprefix(".").removesuffix("()") + + contained = { + edge.get("target") for edge in all_edges if edge.get("relation") == "contains" + } + node_by_id = {node.get("id"): node for node in all_nodes} + + kotlin_scope_by_file: dict[str, dict] = {} + for result in per_file: + scope = result.get("kotlin_scope") + if not scope: + continue + source_file = next( + ( + str(node.get("source_file")) + for node in result.get("nodes", []) + if str(node.get("source_file") or "") + .lower() + .endswith((".kt", ".kts")) + ), + None, + ) + if source_file: + kotlin_scope_by_file[source_file] = scope + + type_def_nids: dict[str, list[str]] = {} + type_def_nids_by_fqn: dict[str, list[str]] = {} + type_fqn_by_nid: dict[str, str] = {} + for node in all_nodes: + if ( + node.get("source_file") + and _lang_family(node.get("source_file")) == "jvm" + and node.get("id") in contained + and _is_type_like_definition(node) + ): + type_name = key(node.get("label", "")) + type_def_nids.setdefault(type_name, []).append(node["id"]) + scope = kotlin_scope_by_file.get(str(node.get("source_file"))) + if scope is not None: + package_name = str(scope.get("package") or "") + qualified_type_name = str( + node.get("_kotlin_qualified_type_name") or type_name + ) + fqn = ( + f"{package_name}.{qualified_type_name}" + if package_name + else qualified_type_name + ) + type_def_nids_by_fqn.setdefault(fqn, []).append(node["id"]) + type_fqn_by_nid[node["id"]] = fqn + + inner_owner_by_type: dict[str, str] = {} + for node in all_nodes: + inner_owner_name = node.get("_kotlin_inner_owner_name") + if not inner_owner_name: + continue + source_file = str(node.get("source_file") or "") + scope = kotlin_scope_by_file.get(source_file) + package_name = str((scope or {}).get("package") or "") + owner_fqn = ( + f"{package_name}.{inner_owner_name}" + if package_name + else str(inner_owner_name) + ) + owner_defs = type_def_nids_by_fqn.get(owner_fqn, []) + if len(owner_defs) == 1: + inner_owner_by_type[node["id"]] = owner_defs[0] + + # A capitalized Kotlin call may name either a constructor or a top-level + # function. Per-file extraction cannot see a competing factory in another + # source file, so constructor-derived receiver facts fail closed when the + # corpus contains a same-named top-level callable. Explicitly typed + # receivers remain safe and are not affected by this guard. + top_level_callable_names: set[str] = set() + top_level_callable_fqns: set[str] = set() + unknown_package_callable_names: set[str] = set() + for node in all_nodes: + source_file = str(node.get("source_file") or "") + if not ( + node.get("id") in contained + and source_file.lower().endswith((".kt", ".kts")) + and not _is_type_like_definition(node) + and str(node.get("label") or "").endswith("()") + ): + continue + callable_name = key(node.get("label", "")) + top_level_callable_names.add(callable_name) + scope = kotlin_scope_by_file.get(source_file) + if scope is None: + unknown_package_callable_names.add(callable_name) + continue + package_name = str(scope.get("package") or "") + top_level_callable_fqns.add( + f"{package_name}.{callable_name}" if package_name else callable_name + ) + for scope in kotlin_scope_by_file.values(): + package_name = str(scope.get("package") or "") + for property_name in scope.get("properties", []): + property_name = str(property_name) + if not property_name: + continue + top_level_callable_names.add(property_name) + top_level_callable_fqns.add( + f"{package_name}.{property_name}" + if package_name + else property_name + ) + + def visible_qualified_fqns( + source_file: str, + local_name: str, + available_fqns: set[str], + ) -> set[str]: + """Return package/import-qualified symbols visible under ``local_name``.""" + scope = kotlin_scope_by_file.get(source_file) + if scope is None: + return set() + + package_name = str(scope.get("package") or "") + local_fqn = f"{package_name}.{local_name}" if package_name else local_name + visible = {local_fqn} if local_fqn in available_fqns else set() + for imported in scope.get("imports", []): + path = str(imported.get("path") or "") + if not path: + continue + if imported.get("star"): + imported_fqn = f"{path}.{local_name}" + if imported_fqn in available_fqns: + visible.add(imported_fqn) + continue + visible_name = str(imported.get("alias") or path.rsplit(".", 1)[-1]) + head, separator, suffix = local_name.partition(".") + if visible_name != head: + continue + imported_fqn = f"{path}.{suffix}" if separator else path + if imported_fqn in available_fqns: + visible.add(imported_fqn) + return visible + + def visible_top_level_callable(source_file: str, local_name: str) -> bool: + """Whether a same-named function/callable property is visible here.""" + if local_name in unknown_package_callable_names: + return True + if source_file not in kotlin_scope_by_file: + return local_name in top_level_callable_names + return bool( + visible_qualified_fqns( + source_file, local_name, top_level_callable_fqns + ) + ) + + def resolve_type_defs(type_name: str, source_file: str) -> list[str]: + """Resolve a Kotlin receiver type through its package/import visibility.""" + if source_file not in kotlin_scope_by_file: + if "." in type_name: + return type_def_nids_by_fqn.get(type_name, []) + return type_def_nids.get(key(type_name), []) + + visible_fqns = visible_qualified_fqns( + source_file, type_name, set(type_def_nids_by_fqn) + ) + if visible_fqns: + return [ + nid + for fqn in visible_fqns + for nid in type_def_nids_by_fqn.get(fqn, []) + ] + if "." in type_name: + return type_def_nids_by_fqn.get(type_name, []) + + # A scoped Kotlin reference that cannot be tied to its own package or an + # explicit import must not fall through to an unrelated corpus-global JVM + # type. Java package/typealias resolution is a separate, richer problem; + # fail closed here instead of fabricating a member edge. + return [] + + method_index: dict[tuple[str, str], set[str]] = {} + enclosing_type: dict[str, str] = {} + direct_supertypes: dict[str, set[str]] = {} + direct_class_supertypes: dict[str, set[str]] = {} + for edge in all_edges: + if edge.get("relation") not in ("inherits", "implements"): + continue + subtype, supertype = edge.get("source"), edge.get("target") + if subtype and supertype: + direct_supertypes.setdefault(subtype, set()).add(supertype) + if edge.get("relation") == "inherits": + direct_class_supertypes.setdefault(subtype, set()).add(supertype) + types_with_supertypes = set(direct_supertypes) + for edge in all_edges: + if edge.get("relation") != "method": + continue + owner, method = edge.get("source"), edge.get("target") + method_node = node_by_id.get(method) + if method_node is None: + continue + enclosing_type.setdefault(method, owner) + method_index.setdefault((owner, key(method_node.get("label", ""))), set()).add( + method + ) + companion_method_index: dict[tuple[str, str], set[str]] = {} + companion_owner_by_method: dict[str, str] = {} + companion_operator_invokes = { + node["id"] + for node in all_nodes + if node.get("_kotlin_companion_operator_invoke") + } + for node in all_nodes: + companion_owner = node.get("_kotlin_companion_owner") + if companion_owner: + companion_owner_by_method[node["id"]] = companion_owner + companion_method_index.setdefault( + (companion_owner, key(node.get("label", ""))), set() + ).add(node["id"]) + + def type_is_or_inherits(type_nid: str | None, ancestor_nid: str) -> bool: + """Whether ``type_nid`` reaches ``ancestor_nid`` in the extracted hierarchy.""" + if not type_nid: + return False + pending = [type_nid] + seen: set[str] = set() + while pending: + current = pending.pop() + if current == ancestor_nid: + return True + if current in seen: + continue + seen.add(current) + pending.extend(direct_supertypes.get(current, ())) + return False + + def has_private_type_access( + caller_owner: str | None, declaring_owner: str + ) -> bool: + """Whether the caller is the declaring type or one of its nested types.""" + if caller_owner == declaring_owner: + return True + caller_fqn = type_fqn_by_nid.get(caller_owner or "") + declaring_fqn = type_fqn_by_nid.get(declaring_owner) + if not caller_fqn or not declaring_fqn: + return False + return ( + caller_fqn.startswith(f"{declaring_fqn}.") + and node_by_id.get(caller_owner, {}).get("source_file") + == node_by_id.get(declaring_owner, {}).get("source_file") + ) + + def owned_methods( + owner_nid: str, method_name: str, caller_owner: str | None + ) -> set[str]: + """Return owned methods visible from the caller's enclosing type.""" + return { + method_nid + for method_nid in method_index.get((owner_nid, method_name), set()) + if not node_by_id.get(method_nid, {}).get("_kotlin_private_function") + or has_private_type_access(caller_owner, owner_nid) + } + + def nearest_methods( + type_nid: str, method_name: str, caller_owner: str | None + ) -> set[str]: + """Return direct or nearest inherited methods, failing closed per level.""" + direct = owned_methods(type_nid, method_name, caller_owner) + if direct: + return direct + class_candidates = nearest_class_methods( + type_nid, method_name, caller_owner + ) + if class_candidates: + return class_candidates + frontier = set(direct_supertypes.get(type_nid, ())) + seen = {type_nid} + while frontier: + candidates = { + method_nid + for owner_nid in frontier + for method_nid in owned_methods( + owner_nid, method_name, caller_owner + ) + } + if candidates: + return candidates + seen.update(frontier) + frontier = { + ancestor + for owner_nid in frontier + for ancestor in direct_supertypes.get(owner_nid, ()) + if ancestor not in seen + } + return set() + + def nearest_class_methods( + type_nid: str, method_name: str, caller_owner: str | None + ) -> set[str]: + """Return the nearest method along Kotlin's concrete class hierarchy.""" + frontier = set(direct_class_supertypes.get(type_nid, ())) + seen = {type_nid} + while frontier: + candidates = { + method_nid + for owner_nid in frontier + for method_nid in owned_methods( + owner_nid, method_name, caller_owner + ) + } + if candidates: + return candidates + seen.update(frontier) + frontier = { + ancestor + for owner_nid in frontier + for ancestor in direct_class_supertypes.get(owner_nid, ()) + if ancestor not in seen + } + return set() + + def nearest_inherited_methods( + type_nid: str, method_name: str, caller_owner: str | None + ) -> set[str]: + """Return methods selected by an explicit ``super`` receiver.""" + return nearest_class_methods(type_nid, method_name, caller_owner) + + extension_index: dict[str, set[str]] = {} + extension_fqn_by_nid: dict[str, str] = {} + extension_fqns: set[str] = set() + for node in all_nodes: + receiver_type = node.get("_kotlin_extension_receiver_type") + if node.get("_kotlin_companion_owner") or not receiver_type or ( + node.get("id") not in contained + and node.get("id") not in enclosing_type + ): + continue + extension_name = key(node.get("label", "")) + source_file = str(node.get("source_file") or "") + scope = kotlin_scope_by_file.get(source_file) + if not extension_name or scope is None: + continue + extension_index.setdefault(extension_name, set()).add(node["id"]) + if node.get("id") in contained: + package_name = str(scope.get("package") or "") + extension_fqn = ( + f"{package_name}.{extension_name}" if package_name else extension_name + ) + extension_fqn_by_nid[node["id"]] = extension_fqn + extension_fqns.add(extension_fqn) + + def extension_matches_type(extension_nid: str, type_nid: str) -> bool: + extension_node = node_by_id.get(extension_nid, {}) + receiver_type = str( + extension_node.get("_kotlin_extension_receiver_type") or "" + ) + type_node = node_by_id.get(type_nid, {}) + type_name = key(type_node.get("label", "")) + if not receiver_type or key(receiver_type).rsplit(".", 1)[-1] != type_name: + return False + type_fqn = type_fqn_by_nid.get(type_nid) + if not type_fqn: + return False + source_file = str(extension_node.get("source_file") or "") + visible_fqns = visible_qualified_fqns( + source_file, receiver_type, set(type_def_nids_by_fqn) + ) + if visible_fqns: + return type_fqn in visible_fqns + return receiver_type == type_fqn + + def extension_is_visible( + extension_nid: str, + caller_nid: str, + caller_source_file: str, + local_name: str, + ) -> bool: + extension_node = node_by_id.get(extension_nid, {}) + dispatch_owner = enclosing_type.get(extension_nid) + if dispatch_owner is not None: + if ( + extension_node.get("_kotlin_private_function") + and not has_private_type_access( + enclosing_type.get(caller_nid), dispatch_owner + ) + ): + return False + caller_owner = enclosing_type.get(caller_nid) + seen_owners: set[str] = set() + while caller_owner and caller_owner not in seen_owners: + if type_is_or_inherits(caller_owner, dispatch_owner): + return True + seen_owners.add(caller_owner) + caller_owner = inner_owner_by_type.get(caller_owner) + return False + extension_source_file = str(extension_node.get("source_file") or "") + if ( + extension_node.get("_kotlin_private_function") + and extension_source_file != caller_source_file + ): + return False + if extension_source_file == caller_source_file: + return True + extension_fqn = extension_fqn_by_nid.get(extension_nid) + return bool( + extension_fqn + and extension_fqn + in visible_qualified_fqns( + caller_source_file, local_name, extension_fqns + ) + ) + + def visible_extension_candidates( + source_file: str, local_name: str + ) -> set[str]: + """Return extensions visible under their declaration name or import alias.""" + candidates = set(extension_index.get(local_name, set())) + scope = kotlin_scope_by_file.get(source_file) + if scope is None: + return candidates + for imported in scope.get("imports", []): + if imported.get("star") or imported.get("alias") != local_name: + continue + path = str(imported.get("path") or "") + declaration_name = path.rsplit(".", 1)[-1] + candidates.update( + extension_nid + for extension_nid in extension_index.get(declaration_name, set()) + if extension_fqn_by_nid.get(extension_nid) == path + ) + return candidates + + def nearest_member_extensions( + candidates: set[str], caller_nid: str + ) -> set[str]: + """Prefer member extensions on the nearest dispatch receiver owner.""" + member_candidates = { + extension_nid + for extension_nid in candidates + if extension_nid in enclosing_type + } + if not member_candidates: + return candidates + caller_owner = enclosing_type.get(caller_nid) + if not caller_owner: + return set() + def select_for_owner(owner_nid: str) -> set[str]: + direct = { + extension_nid + for extension_nid in member_candidates + if enclosing_type.get(extension_nid) == owner_nid + } + if direct: + return direct + + class_frontier = set(direct_class_supertypes.get(owner_nid, ())) + class_seen = {owner_nid} + while class_frontier: + selected = { + extension_nid + for extension_nid in member_candidates + if enclosing_type.get(extension_nid) in class_frontier + } + if selected: + return selected + class_seen.update(class_frontier) + class_frontier = { + ancestor + for current_owner in class_frontier + for ancestor in direct_class_supertypes.get(current_owner, ()) + if ancestor not in class_seen + } + + frontier = set(direct_supertypes.get(owner_nid, ())) + seen = {owner_nid} + while frontier: + selected = { + extension_nid + for extension_nid in member_candidates + if enclosing_type.get(extension_nid) in frontier + } + if selected: + return selected + seen.update(frontier) + frontier = { + ancestor + for current_owner in frontier + for ancestor in direct_supertypes.get(current_owner, ()) + if ancestor not in seen + } + return set() + + seen_owners: set[str] = set() + while caller_owner and caller_owner not in seen_owners: + selected = select_for_owner(caller_owner) + if selected: + return selected + seen_owners.add(caller_owner) + caller_owner = inner_owner_by_type.get(caller_owner) + return set() + + existing_pairs = {(edge.get("source"), edge.get("target")) for edge in all_edges} + for result in per_file: + for raw_call in result.get("raw_calls", []): + if raw_call.get("lang") != "kotlin" or not raw_call.get("is_member_call"): + continue + receiver = raw_call.get("receiver") + callee = raw_call.get("callee") + caller = raw_call.get("caller_nid") + if not callee or not caller: + continue + + qualified_method_nids: set[str] | None = None + exact = ( + receiver == "this" + and not raw_call.get("kotlin_extension_receiver") + ) + if raw_call.get("kotlin_super_receiver"): + type_nid = enclosing_type.get(caller) + if not type_nid: + continue + qualified_method_nids = nearest_inherited_methods( + type_nid, key(callee), enclosing_type.get(caller) + ) + exact = True + elif raw_call.get("kotlin_type_receiver_candidate"): + type_defs = resolve_type_defs( + str(receiver), str(raw_call.get("source_file") or "") + ) + if len(type_defs) != 1: + continue + type_nid = type_defs[0] + if not node_by_id.get(type_nid, {}).get("_kotlin_object_definition"): + qualified_method_nids = companion_method_index.get( + (type_nid, key(callee)), set() + ) + qualified_method_nids = { + method_nid + for method_nid in qualified_method_nids + if not node_by_id.get(method_nid, {}).get( + "_kotlin_private_function" + ) + or has_private_type_access( + enclosing_type.get(caller), + companion_owner_by_method.get(method_nid, ""), + ) + } + exact = True + elif exact: + type_nid = enclosing_type.get(caller) + if not type_nid: + continue + else: + type_name = raw_call.get("receiver_type") + if not type_name: + continue + if ( + raw_call.get("kotlin_constructor_candidate") + and visible_top_level_callable( + str(raw_call.get("source_file") or ""), key(type_name) + ) + ): + continue + type_defs = resolve_type_defs( + str(type_name), str(raw_call.get("source_file") or "") + ) + if len(type_defs) != 1: + continue + type_nid = type_defs[0] + if ( + raw_call.get("kotlin_constructor_candidate") + and ( + node_by_id.get(type_nid, {}).get("_kotlin_object_definition") + or type_nid in companion_operator_invokes + ) + ): + continue + + method_nids = ( + qualified_method_nids + if qualified_method_nids is not None + else nearest_methods( + type_nid, key(callee), enclosing_type.get(caller) + ) + ) + if len(method_nids) > 1: + continue + caller_source_file = str(raw_call.get("source_file") or "") + extension_nids = set() + if qualified_method_nids is None: + extension_nids = nearest_member_extensions({ + extension_nid + for extension_nid in visible_extension_candidates( + caller_source_file, key(callee) + ) + if extension_matches_type(extension_nid, type_nid) + and extension_is_visible( + extension_nid, + caller, + caller_source_file, + key(callee), + ) + }, caller) + # Kotlin member functions outrank extensions only when the member is + # applicable. The graph currently has names but no parameter-arity + # facts, so a same-receiver member/extension pair cannot be selected + # soundly; fail closed instead of manufacturing the member edge. + if method_nids and extension_nids: + continue + if extension_nids and type_nid in types_with_supertypes: + continue + target_nid = next(iter(method_nids), None) + target_exact = exact + if target_nid is None: + if len(extension_nids) != 1: + continue + target_nid = next(iter(extension_nids)) + target_exact = ( + str(node_by_id.get(target_nid, {}).get("source_file") or "") + == caller_source_file + ) + if target_nid == caller or (caller, target_nid) in existing_pairs: + continue + existing_pairs.add((caller, target_nid)) + all_edges.append({ + "source": caller, + "target": target_nid, + "relation": "calls", + "context": "call", + "confidence": "EXTRACTED" if target_exact else "INFERRED", + "confidence_score": 1.0 if target_exact else 0.8, + "source_file": raw_call.get("source_file", ""), + "source_location": raw_call.get("source_location"), + "weight": 1.0, + }) + + def _resolve_objc_member_calls( per_file: list[dict], all_nodes: list[dict], @@ -2831,6 +3476,13 @@ def _key(label: str) -> str: register_language_resolver( LanguageResolver("java_member_calls", frozenset({".java"}), _resolve_java_member_calls) ) +register_language_resolver( + LanguageResolver( + "kotlin_member_calls", + frozenset({".kt", ".kts"}), + _resolve_kotlin_member_calls, + ) +) # Pascal/Delphi cross-file inherited-method-call resolution: a call from a # manual descendant class to a method it inherits from an ancestor declared # in a DIFFERENT file (the common generated-base/manual-descendant split, @@ -4972,6 +5624,13 @@ def _portable_out_of_root_sf(p: Path) -> str: for n in all_nodes: n.pop("origin_file", None) n.pop("_callable", None) # internal indirect_call marker — never ships to graph.json + n.pop("_kotlin_extension_receiver_type", None) + n.pop("_kotlin_object_definition", None) + n.pop("_kotlin_qualified_type_name", None) + n.pop("_kotlin_companion_owner", None) + n.pop("_kotlin_companion_operator_invoke", None) + n.pop("_kotlin_private_function", None) + n.pop("_kotlin_inner_owner_name", None) # Tag AST provenance so the incremental watch rebuild can distinguish # AST-extracted nodes from semantic/LLM nodes. On a full re-extraction diff --git a/graphify/extractors/engine.py b/graphify/extractors/engine.py index edc624a8e..5effe6e9d 100644 --- a/graphify/extractors/engine.py +++ b/graphify/extractors/engine.py @@ -729,6 +729,542 @@ def _kotlin_function_return_type_node(func_node): return c return None + +def _kotlin_function_receiver_type_node(func_node): + """Return the declared receiver type of a Kotlin extension function.""" + name_node = func_node.child_by_field_name("name") + for child in func_node.children: + if child is name_node or child.type in ("identifier", "simple_identifier"): + break + if child.type in ("user_type", "nullable_type", "type_reference"): + return child + return None + + +def _kotlin_type_parameters_in_scope(node, source: bytes) -> set[str]: + """Return Kotlin generic parameter names visible at ``node``.""" + names: set[str] = set() + current = node + while current is not None: + if current.type in ("class_declaration", "object_declaration", "function_declaration"): + for child in current.children: + if child.type != "type_parameters": + continue + for parameter in child.children: + if parameter.type != "type_parameter": + continue + name = next( + ( + _read_text(token, source) + for token in parameter.children + if token.type in ("identifier", "simple_identifier") + ), + None, + ) + if name: + names.add(name) + current = current.parent + return names + + +def _kotlin_receiver_type_name(type_node, source: bytes) -> str | None: + """Return the concrete Kotlin type usable for receiver resolution.""" + if type_node is None: + return None + if type_node.type in ("nullable_type", "parenthesized_type", "type_reference"): + for child in type_node.children: + if child.is_named: + result = _kotlin_receiver_type_name(child, source) + if result: + return result + return None + if type_node.type not in ("user_type", "simple_user_type"): + return None + + identifiers: list[str] = [] + for child in type_node.children: + if child.type in ("identifier", "type_identifier", "simple_identifier"): + identifiers.append(_read_text(child, source)) + elif child.type == "simple_user_type": + nested = _kotlin_receiver_type_name(child, source) + if nested: + identifiers.append(nested) + name = ".".join(identifiers) if identifiers else None + bare_name = identifiers[-1] if identifiers else None + if ( + not name + or bare_name in _KOTLIN_BUILTIN_TYPES + or bare_name in _JAVA_BUILTIN_TYPES + or ( + len(identifiers) == 1 + and bare_name in _kotlin_type_parameters_in_scope(type_node, source) + ) + ): + return None + return name + + +def _kotlin_file_resolution_scope(root_node, source: bytes) -> dict: + """Return package/import facts needed by corpus-level Kotlin resolution.""" + package_name = "" + imports: list[dict[str, str | bool | None]] = [] + property_names: list[str] = [] + + for child in root_node.children: + if child.type == "property_declaration": + property_name = _kotlin_declaration_name(child, source) + if property_name: + property_names.append(property_name) + continue + if child.type == "package_header": + package_node = child.child_by_field_name("path") or next( + ( + part + for part in child.children + if part.is_named + and part.type + in ("qualified_identifier", "identifier", "simple_identifier") + ), + None, + ) + if package_node is not None: + package_name = _read_text(package_node, source) + continue + if child.type not in ("import", "import_header"): + continue + + path_node = child.child_by_field_name("path") or next( + ( + part + for part in child.children + if part.is_named + and part.type + in ("qualified_identifier", "identifier", "simple_identifier") + ), + None, + ) + if path_node is None: + continue + path = _read_text(path_node, source) + if not path: + continue + is_star = any(part.type == "*" for part in child.children) + direct_identifiers = [ + part + for part in child.children + if part.is_named + and part.type in ("identifier", "simple_identifier") + and part is not path_node + ] + alias = ( + _read_text(direct_identifiers[-1], source) + if direct_identifiers + else None + ) + imports.append({"path": path, "alias": alias, "star": is_star}) + + return { + "package": package_name, + "imports": imports, + "properties": property_names, + } + + +def _kotlin_declaration_name(node, source: bytes) -> str | None: + """Return the name declared by a Kotlin property, parameter, or class parameter.""" + for child in node.children: + if child.type in ("identifier", "simple_identifier"): + return _read_text(child, source) or None + if child.type == "variable_declaration": + for nested in child.children: + if nested.type in ("identifier", "simple_identifier"): + return _read_text(nested, source) or None + return None + + +def _kotlin_variable_names(node, source: bytes) -> list[str]: + """Return names introduced by a Kotlin variable or destructuring declaration.""" + names: list[str] = [] + stack = [node] + while stack: + current = stack.pop() + if current.type in ("variable_declaration", "parameter", "class_parameter"): + name = _kotlin_declaration_name(current, source) + if name: + names.append(name) + continue + stack.extend( + child + for child in current.children + if child.type + in ( + "variable_declaration", + "multi_variable_declaration", + "parameter", + "class_parameter", + ) + ) + return names + + +def _kotlin_call_target_name(call_node, source: bytes) -> str | None: + """Return the bare function/constructor named by a simple Kotlin call.""" + if call_node is None or call_node.type != "call_expression": + return None + first = next((child for child in call_node.children if child.is_named), None) + if first is None or first.type not in ("identifier", "simple_identifier"): + return None + return _read_text(first, source) or None + + +def _kotlin_binding_type( + property_node, + source: bytes, + return_types: dict[str, str | None], + shadowed_call_targets: set[str] | frozenset[str] = frozenset(), + allow_implicit_call_inference: bool = True, +) -> tuple[str | None, bool]: + """Infer a Kotlin binding type and whether it came from a constructor guess.""" + explicit = _kotlin_receiver_type_name( + _kotlin_property_type_node(property_node), source + ) + if explicit: + return explicit, False + if not allow_implicit_call_inference: + return None, False + initializer = next( + (child for child in property_node.children if child.type == "call_expression"), + None, + ) + target = _kotlin_call_target_name(initializer, source) + if not target or target in shadowed_call_targets: + return None, False + declared_return = return_types.get(target) + if target[:1].isupper(): + # Kotlin permits a function and a class constructor to share an + # uppercase name. Without argument signatures, choosing either one can + # fabricate a receiver type; retain only unambiguous constructor names. + return (None, False) if target in return_types else (target, True) + return declared_return, False + + +def _kotlin_method_call_target_shadows(method_node, source: bytes) -> set[str]: + """Collect lexical bindings that can shadow getter or constructor calls.""" + shadows: set[str] = set() + + params = next( + ( + child + for child in method_node.children + if child.type == "function_value_parameters" + ), + None, + ) + if params is not None: + for parameter in params.children: + if parameter.type == "parameter": + name = _kotlin_declaration_name(parameter, source) + if name: + shadows.add(name) + + def visit(node) -> None: + if node.type in ("lambda_literal", "anonymous_function"): + return + if node.type in ("function_declaration", "class_declaration", "object_declaration"): + name = _kotlin_declaration_name(node, source) + if name: + shadows.add(name) + return + if node.type in ( + "property_declaration", + "for_statement", + "catch_block", + "when_subject", + ): + shadows.update(_kotlin_variable_names(node, source)) + for child in node.children: + visit(child) + + body = next( + (child for child in method_node.children if child.type == "function_body"), + None, + ) + if body is not None: + visit(body) + return shadows + + +def _kotlin_owner_call_target_shadows(owner_node, source: bytes) -> set[str]: + """Collect callable-valued names visible from functions owned by a file/class.""" + shadows: set[str] = set() + + def add_declaration(node) -> None: + name = _kotlin_declaration_name(node, source) + if name: + shadows.add(name) + + if owner_node.type == "source_file": + for child in owner_node.children: + if child.type == "property_declaration": + add_declaration(child) + return shadows + + for child in owner_node.children: + if child.type == "primary_constructor": + stack = list(child.children) + while stack: + parameter = stack.pop() + if parameter.type == "class_parameter": + if any( + token.type in ("val", "var") + for token in parameter.children + ): + add_declaration(parameter) + continue + stack.extend(parameter.children) + elif child.type == "class_body": + for member in child.children: + if member.type == "property_declaration": + add_declaration(member) + elif member.type == "companion_object": + companion_body = next( + ( + nested + for nested in member.children + if nested.type == "class_body" + ), + None, + ) + if companion_body is not None: + for companion_member in companion_body.children: + if companion_member.type in ( + "function_declaration", + "property_declaration", + ): + add_declaration(companion_member) + return shadows + + +def _kotlin_owner_allows_implicit_call_inference(owner_node) -> bool: + """Whether unqualified calls have no unresolved inherited/outer receiver.""" + if owner_node.type == "source_file": + return True + if any(child.type == "delegation_specifiers" for child in owner_node.children): + return False + current = owner_node.parent + while current is not None: + if current.type in ("class_declaration", "object_declaration"): + return False + current = current.parent + return True + + +def _kotlin_nested_scope_shadow_names(node, source: bytes) -> set[str]: + """Return receiver names introduced by one nested Kotlin lexical scope.""" + names: set[str] = set() + if node.type == "block": + for child in node.children: + if child.type == "property_declaration": + names.update(_kotlin_variable_names(child, source)) + elif node.type == "catch_block": + name = next( + ( + _read_text(child, source) + for child in node.children + if child.type in ("identifier", "simple_identifier") + ), + None, + ) + if name: + names.add(name) + elif node.type == "when_expression": + subject = next( + (child for child in node.children if child.type == "when_subject"), + None, + ) + names.update(_kotlin_variable_names(subject, source)) + elif node.type in ("for_statement", "when_subject"): + names.update(_kotlin_variable_names(node, source)) + return names + + +def _kotlin_class_receiver_types( + class_node, + source: bytes, + return_types: dict[str, str | None], + shadowed_call_targets: set[str], + allow_implicit_call_inference: bool, +) -> tuple[dict[str, str], set[str]]: + """Build the current class's property receiver types conservatively.""" + table: dict[str, str] = {} + ambiguous: set[str] = set() + constructor_inferred: set[str] = set() + + def bind( + name: str | None, + type_name: str | None, + from_constructor: bool = False, + ) -> None: + if not name or not type_name or name in ambiguous: + return + previous = table.get(name) + if previous is not None and previous != type_name: + table.pop(name, None) + constructor_inferred.discard(name) + ambiguous.add(name) + else: + table[name] = type_name + if from_constructor: + constructor_inferred.add(name) + + for child in class_node.children: + if child.type == "primary_constructor": + stack = list(child.children) + while stack: + parameter = stack.pop() + if parameter.type == "class_parameter": + is_property = any( + token.type in ("val", "var") for token in parameter.children + ) + if is_property: + type_node = next( + ( + token + for token in parameter.children + if token.type + in ("user_type", "nullable_type", "type_reference") + ), + None, + ) + bind( + _kotlin_declaration_name(parameter, source), + _kotlin_receiver_type_name(type_node, source), + ) + continue + stack.extend(parameter.children) + elif child.type == "class_body": + for member in child.children: + if member.type == "property_declaration": + type_name, from_constructor = _kotlin_binding_type( + member, + source, + return_types, + shadowed_call_targets, + allow_implicit_call_inference, + ) + bind( + _kotlin_declaration_name(member, source), + type_name, + from_constructor, + ) + + table.update({f"this.{name}": type_name for name, type_name in table.items()}) + constructor_inferred.update( + f"this.{name}" for name in tuple(constructor_inferred) + ) + return table, constructor_inferred + + +def _kotlin_method_receiver_types( + method_node, + source: bytes, + field_types: dict[str, str], + field_constructor_receivers: set[str], + return_types: dict[str, str | None], + shadowed_call_targets: set[str], + allow_implicit_call_inference: bool, +) -> tuple[dict[str, str], set[str]]: + """Build one Kotlin method's receiver table without leaking sibling scopes.""" + receiver_type_node = _kotlin_function_receiver_type_node(method_node) + # An extension function has two implicit receivers. Unqualified properties + # prefer the extension receiver, so treating the dispatch class's fields as + # ordinary locals would bind them to the wrong owner. + table = {} if receiver_type_node is not None else dict(field_types) + constructor_inferred = ( + set() if receiver_type_node is not None else set(field_constructor_receivers) + ) + ambiguous: set[str] = set() + extension_type = _kotlin_receiver_type_name( + receiver_type_node, source + ) + if extension_type: + table["this"] = extension_type + + params = next( + ( + child + for child in method_node.children + if child.type == "function_value_parameters" + ), + None, + ) + if params is not None: + for parameter in params.children: + if parameter.type != "parameter": + continue + name = _kotlin_declaration_name(parameter, source) + type_node = next( + ( + child + for child in parameter.children + if child.type in ("user_type", "nullable_type", "type_reference") + ), + None, + ) + type_name = _kotlin_receiver_type_name(type_node, source) + if name: + table.pop(name, None) + constructor_inferred.discard(name) + if type_name: + table[name] = type_name + else: + ambiguous.add(name) + + def bind_local( + name: str | None, + type_name: str | None, + from_constructor: bool = False, + ) -> None: + if not name or name in ambiguous: + return + previous = table.get(name) + if not type_name or (previous is not None and previous != type_name): + table.pop(name, None) + constructor_inferred.discard(name) + ambiguous.add(name) + else: + table[name] = type_name + if from_constructor: + constructor_inferred.add(name) + + body = next( + (child for child in method_node.children if child.type == "function_body"), + None, + ) + if body is not None: + root_scope = next( + (child for child in body.children if child.type == "block"), body + ) + for statement in root_scope.children: + if statement.type != "property_declaration": + continue + names = _kotlin_variable_names(statement, source) + if len(names) == 1: + type_name, from_constructor = _kotlin_binding_type( + statement, + source, + return_types, + shadowed_call_targets, + allow_implicit_call_inference, + ) + bind_local(names[0], type_name, from_constructor) + else: + for name in names: + bind_local(name, None) + for name in ambiguous: + table.pop(name, None) + return table, constructor_inferred + def _swift_declaration_keyword(node) -> str | None: """Return the leading kind token for a Swift class_declaration: class/struct/enum/extension/actor.""" for c in node.children: @@ -2225,6 +2761,21 @@ def _extract_generic( # while parameters and locals belong only to their declaring method. java_field_types: dict[str, dict[str, str]] = {} java_method_scopes: dict[int, tuple[object, str]] = {} + # Kotlin receiver typing mirrors Java's method-scoped model. Class nodes are + # retained until every same-owner function return type is known, so a property + # or local initialized from a getter can be typed without depending on source order. + kotlin_class_scopes: dict[str, object] = {} + kotlin_method_scopes: dict[int, tuple[object, str]] = {} + kotlin_return_types: dict[str, dict[str, str]] = {} + kotlin_seen_returns: dict[str, set[str]] = {} + kotlin_extension_receiver_types: dict[str, str] = {} + kotlin_object_nids: set[str] = set() + kotlin_qualified_type_names: dict[str, str] = {} + kotlin_class_nids_by_range: dict[tuple[int, int], str] = {} + kotlin_companion_owner_types: dict[str, set[str]] = {} + kotlin_companion_operator_invoke_owners: set[str] = set() + kotlin_private_function_nids: set[str] = set() + kotlin_inner_owner_names: dict[str, str] = {} csharp_interface_names: set[str] = set() if config.ts_module == "tree_sitter_c_sharp": @@ -2362,6 +2913,27 @@ def walk(node, parent_class_nid: str | None = None) -> None: add_node(class_nid, class_name, line, metadata=metadata) callable_def_nids.add(class_nid) # a class is callable (constructor) add_edge(file_nid, class_nid, "contains", line) + if config.ts_module == "tree_sitter_kotlin": + kotlin_class_scopes[class_nid] = node + kotlin_class_nids_by_range[(node.start_byte, node.end_byte)] = class_nid + parent_type_name = kotlin_qualified_type_names.get(parent_class_nid or "") + kotlin_qualified_type_names[class_nid] = ( + f"{parent_type_name}.{class_name}" + if parent_type_name + else class_name + ) + if ( + parent_class_nid + and parent_type_name + and any( + child.type == "modifiers" + and "inner" in _read_text(child, source).split() + for child in node.children + ) + ): + kotlin_inner_owner_names[class_nid] = parent_type_name + if t == "object_declaration": + kotlin_object_nids.add(class_nid) # TS/JS decorators on the class and its members (@Component, @Injectable, # @Input, @Inject, @Entity, …). Decorators live only in class subtrees. @@ -3298,6 +3870,58 @@ def _emit_java_parent_type(type_node, rel: str, at_line: int) -> None: add_edge(func_nid, target_nid, "references", line, context=ctx) if config.ts_module == "tree_sitter_kotlin": + owner_nid = parent_class_nid or file_nid + modifiers = { + modifier + for child in node.children + if child.type == "modifiers" + for modifier in _read_text(child, source).split() + } + is_operator = "operator" in modifiers + if "private" in modifiers: + kotlin_private_function_nids.add(func_nid) + companion_owner = None + current = node.parent + inside_companion = False + while current is not None: + if current.type == "companion_object": + inside_companion = True + elif current.type in ("class_declaration", "object_declaration"): + if inside_companion: + companion_owner = kotlin_class_nids_by_range.get( + (current.start_byte, current.end_byte) + ) + if companion_owner: + kotlin_companion_owner_types.setdefault( + func_nid, set() + ).add(companion_owner) + break + current = current.parent + extension_receiver_type = _kotlin_receiver_type_name( + _kotlin_function_receiver_type_node(node), source + ) + if extension_receiver_type: + kotlin_extension_receiver_types[func_nid] = extension_receiver_type + if ( + companion_owner + and func_name == "invoke" + and is_operator + and not extension_receiver_type + ): + kotlin_companion_operator_invoke_owners.add(companion_owner) + return_type_node = _kotlin_function_return_type_node(node) + receiver_type = _kotlin_receiver_type_name(return_type_node, source) + returns = kotlin_return_types.setdefault(owner_nid, {}) + seen_returns = kotlin_seen_returns.setdefault(owner_nid, set()) + if func_name in seen_returns: + previous = returns.get(func_name) + if not receiver_type or not previous or previous != receiver_type: + returns.pop(func_name, None) + else: + seen_returns.add(func_name) + if receiver_type: + returns[func_name] = receiver_type + params_container = None for c in node.children: if c.type == "function_value_parameters": @@ -3319,7 +3943,6 @@ def _emit_java_parent_type(type_node, rel: str, at_line: int) -> None: target_nid = ensure_named_node(ref_name, line) if target_nid != func_nid: add_edge(func_nid, target_nid, "references", line, context=ctx) - return_type_node = _kotlin_function_return_type_node(node) if return_type_node is not None: refs = [] _kotlin_collect_type_refs(return_type_node, source, False, refs) @@ -3491,6 +4114,11 @@ def _emit_java_parent_type(type_node, rel: str, at_line: int) -> None: if body: if config.ts_module == "tree_sitter_java" and parent_class_nid: java_method_scopes[id(body)] = (node, parent_class_nid) + elif config.ts_module == "tree_sitter_kotlin": + kotlin_method_scopes[id(body)] = ( + node, + parent_class_nid or file_nid, + ) function_bodies.append((func_nid, body)) return @@ -3598,7 +4226,87 @@ def _emit_java_parent_type(type_node, rel: str, at_line: int) -> None: ) for body_id, (method_node, class_nid) in java_method_scopes.items() } - + kotlin_return_facts = { + owner_nid: { + name: kotlin_return_types.get(owner_nid, {}).get(name) + for name in names + } + for owner_nid, names in kotlin_seen_returns.items() + } + kotlin_field_types: dict[str, dict[str, str]] = {} + kotlin_field_constructor_receivers: dict[str, set[str]] = {} + kotlin_owner_call_target_shadows = { + file_nid: _kotlin_owner_call_target_shadows(root, source), + **{ + class_nid: _kotlin_owner_call_target_shadows(class_node, source) + for class_nid, class_node in kotlin_class_scopes.items() + }, + } + kotlin_owner_implicit_call_inference = { + file_nid: True, + **{ + class_nid: _kotlin_owner_allows_implicit_call_inference(class_node) + for class_nid, class_node in kotlin_class_scopes.items() + }, + } + for class_nid, class_node in kotlin_class_scopes.items(): + field_types, constructor_receivers = _kotlin_class_receiver_types( + class_node, + source, + kotlin_return_facts.get(class_nid, {}), + kotlin_owner_call_target_shadows.get(class_nid, set()), + kotlin_owner_implicit_call_inference.get(class_nid, False), + ) + kotlin_field_types[class_nid] = field_types + kotlin_field_constructor_receivers[class_nid] = constructor_receivers + + kotlin_receiver_types: dict[int, dict[str, str]] = {} + kotlin_constructor_receivers: dict[int, set[str]] = {} + kotlin_call_target_shadows = { + body_id: ( + kotlin_owner_call_target_shadows.get(owner_nid, set()) + | _kotlin_method_call_target_shadows(method_node, source) + ) + for body_id, (method_node, owner_nid) in kotlin_method_scopes.items() + } + kotlin_implicit_call_inference = { + body_id: ( + kotlin_owner_implicit_call_inference.get(owner_nid, False) + and _kotlin_function_receiver_type_node(method_node) is None + ) + for body_id, (method_node, owner_nid) in kotlin_method_scopes.items() + } + for body_id, (method_node, owner_nid) in kotlin_method_scopes.items(): + receiver_types, constructor_receivers = _kotlin_method_receiver_types( + method_node, + source, + kotlin_field_types.get(owner_nid, {}), + kotlin_field_constructor_receivers.get(owner_nid, set()), + kotlin_return_facts.get(owner_nid, {}), + kotlin_call_target_shadows.get(body_id, set()), + kotlin_implicit_call_inference.get(body_id, False), + ) + kotlin_receiver_types[body_id] = receiver_types + kotlin_constructor_receivers[body_id] = constructor_receivers + kotlin_method_returns = { + body_id: kotlin_return_facts.get(owner_nid, {}) + for body_id, (_method_node, owner_nid) in kotlin_method_scopes.items() + } + kotlin_extension_receivers = { + body_id: _kotlin_function_receiver_type_node(method_node) is not None + for body_id, (method_node, _owner_nid) in kotlin_method_scopes.items() + } + kotlin_root_scope_ids = { + body_id: id( + next( + (child for child in body_node.children if child.type == "block"), + body_node, + ) + ) + for _caller_nid, body_node in function_bodies + for body_id in (id(body_node),) + if body_id in kotlin_method_scopes + } def _emit_indirect_by_name(ident_name: str, loc_node, scope_nid: str, context: str) -> None: """Resolve a name that is referenced AS A VALUE to a real callable def and emit @@ -3744,6 +4452,15 @@ def walk_calls( node, caller_nid: str, java_types: dict[str, str] | None = None, + kotlin_types: dict[str, str] | None = None, + kotlin_returns: dict[str, str | None] | None = None, + kotlin_nested_callable: bool = False, + kotlin_extension_receiver: bool = False, + kotlin_constructor_receiver_names: set[str] | None = None, + kotlin_shadowed_call_targets: set[str] | None = None, + kotlin_allow_implicit_call_inference: bool = True, + kotlin_root_scope_id: int | None = None, + kotlin_lexical_shadows: frozenset[str] = frozenset(), ) -> None: if node.type in config.function_boundary_types: # JS/TS: an inline/returned closure not separately tracked in @@ -3757,7 +4474,20 @@ def walk_calls( body = node.child_by_field_name("body") if body is not None and id(body) not in _tracked_body_ids: for child in node.children: - walk_calls(child, caller_nid, java_types) + walk_calls( + child, + caller_nid, + java_types, + kotlin_types, + kotlin_returns, + kotlin_nested_callable, + kotlin_extension_receiver, + kotlin_constructor_receiver_names, + kotlin_shadowed_call_targets, + kotlin_allow_implicit_call_inference, + kotlin_root_scope_id, + kotlin_lexical_shadows, + ) return if node.type in config.call_types: @@ -3767,7 +4497,20 @@ def walk_calls( edges, seen_dyn_import_pairs): # Still recurse into children (import().then(...) may have calls) for child in node.children: - walk_calls(child, caller_nid, java_types) + walk_calls( + child, + caller_nid, + java_types, + kotlin_types, + kotlin_returns, + kotlin_nested_callable, + kotlin_extension_receiver, + kotlin_constructor_receiver_names, + kotlin_shadowed_call_targets, + kotlin_allow_implicit_call_inference, + kotlin_root_scope_id, + kotlin_lexical_shadows, + ) return callee_name: str | None = None @@ -3775,6 +4518,11 @@ def walk_calls( is_this_field_call: bool = False swift_receiver: str | None = None member_receiver: str | None = None + member_receiver_type: str | None = None + member_receiver_from_constructor = False + kotlin_super_call = False + kotlin_type_receiver_candidate = False + kotlin_ambiguous_receiver = False # Special handling per language if config.ts_module == "tree_sitter_swift": @@ -3810,6 +4558,87 @@ def walk_calls( if child.type in ("simple_identifier", "identifier"): callee_name = _read_text(child, source) break + if not kotlin_nested_callable: + receiver = first.children[0] if first.children else None + if receiver is not None and receiver.type in ( + "simple_identifier", + "identifier", + ): + member_receiver = _read_text(receiver, source) + kotlin_type_receiver_candidate = bool( + member_receiver[:1].isupper() + and member_receiver + not in (kotlin_shadowed_call_targets or set()) + ) + elif ( + receiver is not None + and receiver.type == "this_expression" + and _read_text(receiver, source) == "this" + ): + member_receiver = "this" + elif ( + receiver is not None + and receiver.type == "this_expression" + ): + kotlin_ambiguous_receiver = True + elif ( + receiver is not None + and receiver.type == "super_expression" + and _read_text(receiver, source) == "super" + ): + member_receiver = "super" + kotlin_super_call = True + elif ( + receiver is not None + and receiver.type == "navigation_expression" + ): + parts = [ + child for child in receiver.children if child.is_named + ] + if ( + len(parts) == 2 + and parts[0].type == "this_expression" + and _read_text(parts[0], source) == "this" + and parts[1].type + in ("simple_identifier", "identifier") + ): + member_receiver = ( + f"this.{_read_text(parts[1], source)}" + ) + elif receiver is not None and receiver.type == "call_expression": + kotlin_ambiguous_receiver = True + target = _kotlin_call_target_name(receiver, source) + if ( + kotlin_allow_implicit_call_inference + and target + and target not in ( + kotlin_shadowed_call_targets or set() + ) + ): + return_facts = kotlin_returns or {} + declared_return = return_facts.get(target) + if target[:1].isupper(): + member_receiver_type = ( + None if target in return_facts else target + ) + member_receiver_from_constructor = ( + member_receiver_type is not None + ) + else: + member_receiver_type = declared_return + if member_receiver and member_receiver_type is None: + if not ( + kotlin_extension_receiver + and member_receiver.startswith("this.") + ) and member_receiver not in kotlin_lexical_shadows: + member_receiver_type = (kotlin_types or {}).get( + member_receiver + ) + member_receiver_from_constructor = ( + member_receiver_type is not None + and member_receiver + in (kotlin_constructor_receiver_names or set()) + ) elif config.ts_module == "tree_sitter_scala": # Scala: first child first = node.children[0] if node.children else None @@ -4012,7 +4841,20 @@ def walk_calls( _java_defer = ( config.ts_module == "tree_sitter_java" and is_member_call ) - if _java_defer or ( + _kotlin_defer = ( + config.ts_module == "tree_sitter_kotlin" + and is_member_call + and ( + member_receiver_type is not None + or kotlin_nested_callable + or kotlin_ambiguous_receiver + or kotlin_type_receiver_candidate + or kotlin_super_call + or member_receiver + in (kotlin_shadowed_call_targets or set()) + ) + ) + if _java_defer or _kotlin_defer or ( is_member_call and member_receiver and ( @@ -4071,6 +4913,18 @@ def walk_calls( receiver_type = (java_types or {}).get(member_receiver or "") if receiver_type: rc_entry["receiver_type"] = receiver_type + if config.ts_module == "tree_sitter_kotlin": + rc_entry["lang"] = "kotlin" + if kotlin_extension_receiver: + rc_entry["kotlin_extension_receiver"] = True + if member_receiver_type: + rc_entry["receiver_type"] = member_receiver_type + if member_receiver_from_constructor: + rc_entry["kotlin_constructor_candidate"] = True + if kotlin_type_receiver_candidate: + rc_entry["kotlin_type_receiver_candidate"] = True + if kotlin_super_call: + rc_entry["kotlin_super_receiver"] = True raw_calls.append(rc_entry) # Indirect dispatch: a function passed BY NAME as a call argument @@ -4277,7 +5131,33 @@ def walk_calls( _emit_indirect_ref(ident, caller_nid, enclosing_locals, "return") for child in node.children: - walk_calls(child, caller_nid, java_types) + nested_callable = kotlin_nested_callable or ( + config.ts_module == "tree_sitter_kotlin" + and child.type in ("lambda_literal", "anonymous_function") + ) + lexical_shadows = kotlin_lexical_shadows + if ( + config.ts_module == "tree_sitter_kotlin" + and id(child) != kotlin_root_scope_id + ): + lexical_shadows = frozenset( + set(lexical_shadows) + | _kotlin_nested_scope_shadow_names(child, source) + ) + walk_calls( + child, + caller_nid, + java_types, + kotlin_types, + kotlin_returns, + nested_callable, + kotlin_extension_receiver, + kotlin_constructor_receiver_names, + kotlin_shadowed_call_targets, + kotlin_allow_implicit_call_inference, + kotlin_root_scope_id, + lexical_shadows, + ) if config.ts_module == "tree_sitter_ruby": for caller_nid, body_node in function_bodies: @@ -4311,6 +5191,14 @@ def walk_calls( body_node, caller_nid, java_receiver_types.get(id(body_node)), + kotlin_receiver_types.get(id(body_node)), + kotlin_method_returns.get(id(body_node)), + False, + kotlin_extension_receivers.get(id(body_node), False), + kotlin_constructor_receivers.get(id(body_node)), + kotlin_call_target_shadows.get(id(body_node)), + kotlin_implicit_call_inference.get(id(body_node), True), + kotlin_root_scope_ids.get(id(body_node)), ) # #1356: walk property/field initializers (collected above). walk_calls @@ -4418,8 +5306,42 @@ def _scan_js_module_dispatch(n) -> None: for n in nodes: if n["id"] in callable_def_nids: n["_callable"] = True + if kotlin_extension_receiver_types: + for n in nodes: + receiver_type = kotlin_extension_receiver_types.get(n["id"]) + if receiver_type: + n["_kotlin_extension_receiver_type"] = receiver_type + if kotlin_object_nids: + for n in nodes: + if n["id"] in kotlin_object_nids: + n["_kotlin_object_definition"] = True + if kotlin_qualified_type_names: + for n in nodes: + qualified_name = kotlin_qualified_type_names.get(n["id"]) + if qualified_name: + n["_kotlin_qualified_type_name"] = qualified_name + if kotlin_companion_owner_types: + for n in nodes: + companion_owners = kotlin_companion_owner_types.get(n["id"], set()) + if len(companion_owners) == 1: + n["_kotlin_companion_owner"] = next(iter(companion_owners)) + if kotlin_companion_operator_invoke_owners: + for n in nodes: + if n["id"] in kotlin_companion_operator_invoke_owners: + n["_kotlin_companion_operator_invoke"] = True + if kotlin_private_function_nids: + for n in nodes: + if n["id"] in kotlin_private_function_nids: + n["_kotlin_private_function"] = True + if kotlin_inner_owner_names: + for n in nodes: + inner_owner = kotlin_inner_owner_names.get(n["id"]) + if inner_owner: + n["_kotlin_inner_owner_name"] = inner_owner if swift_extensions: result["swift_extensions"] = swift_extensions + if config.ts_module == "tree_sitter_kotlin": + result["kotlin_scope"] = _kotlin_file_resolution_scope(root, source) # TS/JS: augment the constructor-injection type table with local `new` # bindings and type-annotated parameters, so `const s = new Svc(); s.m()` and # a call on a typed param (incl. inside a closure) resolve (#1630). The diff --git a/tests/test_kotlin_member_calls.py b/tests/test_kotlin_member_calls.py new file mode 100644 index 000000000..7766a4b39 --- /dev/null +++ b/tests/test_kotlin_member_calls.py @@ -0,0 +1,1272 @@ +"""Kotlin receiver-typed member-call resolution (#1699).""" + +from __future__ import annotations + +from pathlib import Path + +from graphify.extract import extract + + +def _extract(tmp_path: Path, files: dict[str, str]) -> dict: + paths = [] + for name, body in files.items(): + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body, encoding="utf-8") + paths.append(path) + return extract(paths, cache_root=tmp_path / "graphify-out", parallel=False) + + +def _find(result: dict, label: str, id_contains: str) -> str: + return next( + node["id"] + for node in result["nodes"] + if node.get("label") == label and id_contains in node["id"] + ) + + +def _call_edges(result: dict) -> list[dict]: + return [edge for edge in result["edges"] if edge.get("relation") == "calls"] + + +def test_issue_1699_resolves_four_typed_receiver_shapes(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "InputView.kt": ( + "class InputView {\n" + " fun updateKeyboardShow(show: Boolean) {}\n" + "}\n" + ), + "PanelController.kt": ( + "class PanelController {\n" + " private val input = InputView()\n" + " fun getInputView(): InputView = input\n" + " fun onPanelClose() { input.updateKeyboardShow(false) }\n" + " fun onPanelOpen() { getInputView().updateKeyboardShow(true) }\n" + " fun onPanelToggle() {\n" + " val view = getInputView()\n" + " view.updateKeyboardShow(true)\n" + " }\n" + "}\n" + ), + "Window.kt": ( + "fun open() {\n" + " val view = InputView()\n" + " view.updateKeyboardShow(true)\n" + "}\n" + ), + }) + + update = _find(result, ".updateKeyboardShow()", "inputview") + callers = { + _find(result, ".onPanelClose()", "panelcontroller"), + _find(result, ".onPanelOpen()", "panelcontroller"), + _find(result, ".onPanelToggle()", "panelcontroller"), + _find(result, "open()", "window"), + } + resolved = { + edge["source"]: edge + for edge in _call_edges(result) + if edge.get("target") == update and edge.get("source") in callers + } + assert set(resolved) == callers + assert all(edge.get("confidence") == "INFERRED" for edge in resolved.values()) + assert all(edge.get("confidence_score") == 0.8 for edge in resolved.values()) + + get_input = _find(result, ".getInputView()", "panelcontroller") + assert any( + edge["source"] in callers + and edge["target"] == get_input + and edge.get("confidence") == "EXTRACTED" + for edge in _call_edges(result) + ) + + +def test_typed_receiver_selects_its_owner_not_same_named_local_method( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun updateKeyboardShow(show: Boolean) {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun updateKeyboardShow(show: Boolean) {}\n" + " fun run() { input.updateKeyboardShow(true) }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + expected = _find(result, ".updateKeyboardShow()", "inputview") + wrong = _find(result, ".updateKeyboardShow()", "caller") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_ambiguous_receiver_type_emits_no_member_edge(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": "package a\nclass InputView { fun updateKeyboardShow() {} }\n", + "b/InputView.kt": "package b\nclass InputView { fun updateKeyboardShow() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun run() { input.updateKeyboardShow() }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + assert not any( + edge["source"] == run + and "updatekeyboardshow" in str(edge["target"]).lower() + for edge in _call_edges(result) + ) + + +def test_receiver_bindings_are_scoped_per_method(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Alpha.kt": "class Alpha { fun act() {} }\n", + "Beta.kt": "class Beta { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " fun alpha() { val service = Alpha(); service.act() }\n" + " fun beta() { val service = Beta(); service.act() }\n" + "}\n" + ), + }) + + alpha = _find(result, ".alpha()", "caller") + beta = _find(result, ".beta()", "caller") + alpha_act = _find(result, ".act()", "alpha") + beta_act = _find(result, ".act()", "beta") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (alpha, alpha_act) in calls + assert (alpha, beta_act) not in calls + assert (beta, beta_act) in calls + assert (beta, alpha_act) not in calls + + +def test_parameter_shadow_does_not_reuse_field_type_but_this_field_does( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun updateKeyboardShow() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun shadowed(input: Any) { input.updateKeyboardShow() }\n" + " fun explicit(input: Any) { this.input.updateKeyboardShow() }\n" + " fun helper() {}\n" + " fun unqualified() { helper() }\n" + "}\n" + ), + }) + + shadowed = _find(result, ".shadowed()", "caller") + explicit = _find(result, ".explicit()", "caller") + unqualified = _find(result, ".unqualified()", "caller") + helper = _find(result, ".helper()", "caller") + update = _find(result, ".updateKeyboardShow()", "inputview") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (shadowed, update) not in calls + assert (explicit, update) in calls + assert (unqualified, helper) in calls + + +def test_lexical_binders_do_not_reuse_same_named_field_types(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val item = InputView()\n" + " private val error = InputView()\n" + " private val input = InputView()\n" + " private val explicit = InputView()\n" + " private val it = InputView()\n" + " private val anonymousInput = InputView()\n" + " private val whenInput = InputView()\n" + " fun loop(items: List) { for (item in items) { item.act() } }\n" + " fun caught() {\n" + " try { println(\"x\") } catch (error: Exception) { error.act() }\n" + " }\n" + " fun destructured(box: Pair) {\n" + " val (unused, input) = box\n" + " input.act()\n" + " }\n" + " fun explicitLambda(items: List) {\n" + " items.forEach { explicit -> explicit.act() }\n" + " }\n" + " fun implicitLambda(items: List) { items.forEach { it.act() } }\n" + " fun anonymous() {\n" + " val block = fun(anonymousInput: Any) { anonymousInput.act() }\n" + " }\n" + " fun whenBound(other: Any) {\n" + " when (val whenInput = other) { else -> whenInput.act() }\n" + " }\n" + "}\n" + ), + }) + + act = _find(result, ".act()", "inputview") + callers = { + _find(result, f".{name}()", "caller") + for name in ( + "loop", + "caught", + "destructured", + "explicitLambda", + "implicitLambda", + "anonymous", + "whenBound", + ) + } + assert not any( + edge["source"] in callers and edge["target"] == act + for edge in _call_edges(result) + ) + + +def test_type_parameters_do_not_resolve_to_same_named_concrete_class( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "GenericCaller.kt": ( + "class GenericCaller(private val input: InputView) {\n" + " fun run() { input.act() }\n" + "}\n" + ), + "FunctionCaller.kt": ( + "class FunctionCaller {\n" + " fun run(input: InputView) { input.act() }\n" + "}\n" + ), + }) + + act = _find(result, ".act()", "inputview") + callers = { + _find(result, ".run()", "genericcaller"), + _find(result, ".run()", "functioncaller"), + } + assert not any( + edge["source"] in callers and edge["target"] == act + for edge in _call_edges(result) + ) + + +def test_uppercase_factory_constructor_collision_is_left_unresolved( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView(val name: String) { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " fun InputView(value: Int): Other = Other()\n" + " fun factory() { val view = InputView(1); view.act() }\n" + " fun constructor() { val view = InputView(\"real\"); view.act() }\n" + " fun chained() { InputView(1).act() }\n" + "}\n" + ), + }) + + callers = { + _find(result, f".{name}()", "caller") + for name in ("factory", "constructor", "chained") + } + input_act = _find(result, ".act()", "inputview") + other_act = _find(result, ".act()", "other") + assert not any( + edge["source"] in callers and edge["target"] in {input_act, other_act} + for edge in _call_edges(result) + ) + + +def test_top_level_factory_constructor_collision_is_left_unresolved( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView(val name: String) { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Factory.kt": "fun InputView(value: Int): Other = Other()\n", + "Caller.kt": ( + "class Caller {\n" + " fun factory() { val view = InputView(1); view.act() }\n" + " fun constructor() { val view = InputView(\"real\"); view.act() }\n" + " fun chained() { InputView(1).act() }\n" + " fun explicit(view: InputView) { view.act() }\n" + "}\n" + ), + }) + + callers = { + _find(result, f".{name}()", "caller") + for name in ("factory", "constructor", "chained") + } + input_act = _find(result, ".act()", "inputview") + other_act = _find(result, ".act()", "other") + assert not any( + edge["source"] in callers and edge["target"] in {input_act, other_act} + for edge in _call_edges(result) + ) + explicit = _find(result, ".explicit()", "caller") + assert any( + edge["source"] == explicit and edge["target"] == input_act + for edge in _call_edges(result) + ) + + +def test_same_file_extension_member_call_keeps_direct_edge(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Extensions.kt": ( + "class InputView\n" + "fun InputView.decorate() {}\n" + "fun run(view: InputView) { view.decorate() }\n" + ), + }) + + run = _find(result, "run()", "extensions") + decorate = _find(result, "decorate()", "extensions") + edge = next( + edge + for edge in _call_edges(result) + if edge["source"] == run and edge["target"] == decorate + ) + assert edge["confidence"] == "EXTRACTED" + + +def test_constructor_provenance_survives_disjoint_same_named_bindings( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView(val name: String) { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Factory.kt": "fun InputView(value: Int): Other = Other()\n", + "Caller.kt": ( + "class Caller {\n" + " fun getInputView(): InputView = InputView(\"real\")\n" + " fun run(flag: Boolean) {\n" + " if (flag) { val view = InputView(1); view.act() }\n" + " else { val view = getInputView() }\n" + " }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + input_act = _find(result, ".act()", "inputview") + other_act = _find(result, ".act()", "other") + assert not any( + edge["source"] == run and edge["target"] in {input_act, other_act} + for edge in _call_edges(result) + ) + + +def test_extension_this_uses_declared_receiver_not_enclosing_class( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputValue.kt": "class InputValue { fun act() {} }\n", + "InputView.kt": "class InputView { val input = InputValue(); fun act() {} }\n", + "CallerValue.kt": "class CallerValue { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " val input = CallerValue()\n" + " fun act() {}\n" + " fun InputView.extensionRun() { this.act() }\n" + " fun InputView.genericRun() { this.act() }\n" + " fun InputView.propertyRun() { input.act() }\n" + "}\n" + ), + }) + + extension_run = _find(result, ".extensionRun()", "caller") + generic_run = _find(result, ".genericRun()", "caller") + property_run = _find(result, ".propertyRun()", "caller") + caller_act = _find(result, ".act()", "caller") + caller_value_act = _find(result, ".act()", "callervalue") + input_act = _find(result, ".act()", "inputview") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (extension_run, input_act) in calls + assert (extension_run, caller_act) not in calls + assert (generic_run, input_act) not in calls + assert (generic_run, caller_act) not in calls + assert (property_run, caller_value_act) not in calls + + +def test_receiver_typed_calls_inside_lambdas_are_left_unresolved( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Other.kt": ( + "class OtherInput { fun act() {} }\n" + "class Other { val input = OtherInput(); fun act() {} }\n" + ), + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun act() {}\n" + " fun run(other: Other) { with(other) { this.act(); input.act() } }\n" + " fun direct() { this.act() }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + direct = _find(result, ".direct()", "caller") + caller_act = _find(result, ".act()", "caller") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert not any(source == run and "act" in target for source, target in calls) + assert (direct, caller_act) in calls + + +def test_overloaded_getter_with_different_return_types_is_ambiguous( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun isEmpty() = false }\n", + "Caller.kt": ( + "class Caller {\n" + " fun get(value: Int): InputView = InputView()\n" + " fun get(value: String): String = value\n" + " fun local() { val view = get(\"x\"); view.isEmpty() }\n" + " fun chained() { get(\"x\").isEmpty() }\n" + "}\n" + ), + }) + + callers = { + _find(result, ".local()", "caller"), + _find(result, ".chained()", "caller"), + } + input_empty = _find(result, ".isEmpty()", "inputview") + assert not any( + edge["source"] in callers and edge["target"] == input_empty + for edge in _call_edges(result) + ) + + +def test_lexical_callables_shadow_getter_and_constructor_inference( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " fun get(): InputView = InputView()\n" + " fun localFactory() {\n" + " fun InputView(): Other = Other()\n" + " val value = InputView()\n" + " value.act()\n" + " InputView().act()\n" + " }\n" + " fun callableParameter(get: () -> Other) {\n" + " val value = get()\n" + " value.act()\n" + " get().act()\n" + " }\n" + "}\n" + ), + }) + + callers = { + _find(result, ".localFactory()", "caller"), + _find(result, ".callableParameter()", "caller"), + } + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] in callers and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_lambda_shadow_does_not_poison_later_outer_receiver_call( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun run(items: List) {\n" + " items.forEach { input -> input.toString() }\n" + " input.act()\n" + " }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + act = _find(result, ".act()", "inputview") + assert any( + edge["source"] == run and edge["target"] == act + for edge in _call_edges(result) + ) + + +def test_inner_block_shadow_does_not_poison_later_outer_receiver_call( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " private val input = InputView()\n" + " fun run(flag: Boolean) {\n" + " if (flag) {\n" + " val input = Other()\n" + " input.act()\n" + " }\n" + " input.act()\n" + " }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + expected = _find(result, ".act()", "inputview") + wrong = _find(result, ".act()", "other") + calls = [ + edge + for edge in _call_edges(result) + if edge["source"] == run and edge["target"] in {expected, wrong} + ] + assert [edge["target"] for edge in calls] == [expected] + + +def test_labeled_this_member_calls_are_left_unresolved(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Outer.kt": ( + "class Outer {\n" + " fun act() {}\n" + " inner class Inner {\n" + " fun act() {}\n" + " fun run() { this@Outer.act() }\n" + " }\n" + "}\n" + ), + "Caller.kt": ( + "class Caller {\n" + " fun act() {}\n" + " fun InputView.runExtension() { this@Caller.act() }\n" + "}\n" + ), + }) + + callers = { + _find(result, ".run()", "inner"), + _find(result, ".runExtension()", "caller"), + } + acts = { + _find(result, ".act()", "outer"), + _find(result, ".act()", "inner"), + _find(result, ".act()", "caller"), + _find(result, ".act()", "inputview"), + } + assert not any( + edge["source"] in callers and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_qualified_receiver_type_selects_exact_package(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": "package a\nclass InputView { fun act() {} }\n", + "b/InputView.kt": "package b\nclass InputView { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " fun run(view: a.InputView) { view.act() }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + expected = _find(result, ".act()", "a_inputview") + wrong = _find(result, ".act()", "b_inputview") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_unrelated_package_factory_does_not_hide_local_constructor( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": ( + "package a\n" + "class InputView { fun act() {} }\n" + ), + "a/Caller.kt": ( + "package a\n" + "class Caller { fun run() { val view = InputView(); view.act() } }\n" + ), + "b/Factory.kt": ( + "package b\n" + "class Other { fun act() {} }\n" + "fun InputView(): Other = Other()\n" + ), + }) + + run = _find(result, ".run()", "caller") + expected = _find(result, ".act()", "a_inputview") + wrong = _find(result, ".act()", "b_factory_other") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_imported_factory_collision_remains_unresolved(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": ( + "package a\n" + "class InputView { fun act() {} }\n" + ), + "a/Caller.kt": ( + "package a\n" + "import b.InputView\n" + "class Caller { fun run() { val view = InputView(); view.act() } }\n" + ), + "b/Factory.kt": ( + "package b\n" + "class Other { fun act() {} }\n" + "fun InputView(): Other = Other()\n" + ), + }) + + run = _find(result, ".run()", "caller") + acts = { + _find(result, ".act()", "a_inputview"), + _find(result, ".act()", "b_factory_other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_wildcard_and_alias_imported_factories_remain_unresolved( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": ( + "package a\n" + "class InputView { fun act() {} }\n" + ), + "a/StarCaller.kt": ( + "package a\n" + "import b.*\n" + "class StarCaller { fun run() { val view = InputView(); view.act() } }\n" + ), + "a/AliasCaller.kt": ( + "package a\n" + "import c.Factory as InputView\n" + "class AliasCaller { fun run() { val view = InputView(); view.act() } }\n" + ), + "b/Factory.kt": ( + "package b\n" + "class Other { fun act() {} }\n" + "fun InputView(): Other = Other()\n" + ), + "c/Factory.kt": ( + "package c\n" + "class Another { fun act() {} }\n" + "fun Factory(): Another = Another()\n" + ), + }) + + callers = { + _find(result, ".run()", "starcaller"), + _find(result, ".run()", "aliascaller"), + } + acts = { + _find(result, ".act()", "a_inputview"), + _find(result, ".act()", "b_factory_other"), + _find(result, ".act()", "c_factory_another"), + } + assert not any( + edge["source"] in callers and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_import_alias_resolves_receiver_to_imported_type(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": "package a\nclass InputView { fun act() {} }\n", + "b/View.kt": "package b\nclass View { fun act() {} }\n", + "c/Caller.kt": ( + "package c\n" + "import a.InputView as View\n" + "class Caller { fun run(view: View) { view.act() } }\n" + ), + }) + + run = _find(result, ".run()", "caller") + expected = _find(result, ".act()", "a_inputview") + wrong = _find(result, ".act()", "b_view") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_unresolved_typealias_does_not_fall_back_to_unrelated_java_type( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": "package a\nclass InputView { fun act() {} }\n", + "b/View.java": ( + "package b;\n" + "public class View { public void act() {} }\n" + ), + "c/Caller.kt": ( + "package c\n" + "typealias View = a.InputView\n" + "class Caller { fun run(view: View) { view.act() } }\n" + ), + }) + + run = _find(result, ".run()", "caller") + wrong = _find(result, ".act()", "b_view") + assert not any( + edge["source"] == run and edge["target"] == wrong + for edge in _call_edges(result) + ) + + +def test_callable_class_property_shadows_constructor_inference( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Caller.kt": ( + "class Caller {\n" + " val InputView: () -> Other = { Other() }\n" + " fun run() {\n" + " val value = InputView()\n" + " value.act()\n" + " InputView().act()\n" + " }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_same_package_callable_property_shadows_imported_constructor( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "a/InputView.kt": ( + "package a\n" + "class InputView { fun act() {} }\n" + ), + "c/Factory.kt": ( + "package c\n" + "class Other { fun act() {} }\n" + "val InputView: () -> Other = { Other() }\n" + ), + "c/Caller.kt": ( + "package c\n" + "import a.*\n" + "class Caller { fun run() { val value = InputView(); value.act() } }\n" + ), + }) + + run = _find(result, ".run()", "caller") + acts = { + _find(result, ".act()", "a_inputview"), + _find(result, ".act()", "c_factory_other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_callable_object_is_not_treated_as_constructor(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "class Other { fun act() {} }\n" + "object InputView {\n" + " operator fun invoke(): Other = Other()\n" + " fun act() {}\n" + "}\n" + "fun run() { val value = InputView(); value.act() }\n" + ), + }) + + run = _find(result, "run()", "calls") + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_only_companion_operator_invoke_shadows_constructor(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Operator.kt": ( + "class Other { fun act() {} }\n" + "class InputView private constructor(value: Int) {\n" + " companion object {\n" + " operator fun invoke(): Other = Other()\n" + " }\n" + " fun act() {}\n" + "}\n" + "fun run() { val value = InputView(); value.act() }\n" + ), + "Plain.kt": ( + "class PlainView {\n" + " companion object { fun invoke() = Unit }\n" + " fun act() {}\n" + "}\n" + "fun plain() { val value = PlainView(); value.act() }\n" + ), + }) + + run = _find(result, "run()", "operator") + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + plain = _find(result, "plain()", "plain") + plain_act = _find(result, ".act()", "plainview") + assert any( + edge["source"] == plain and edge["target"] == plain_act + for edge in _call_edges(result) + ) + + +def test_unresolved_implicit_receivers_block_constructor_inference( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Other.kt": "class Other { fun act() {} }\n", + "Inherited.kt": ( + "open class Base { val InputView: () -> Other = { Other() } }\n" + "class Caller : Base() {\n" + " fun run() { val value = InputView(); value.act() }\n" + "}\n" + ), + "Extension.kt": ( + "class Scope { val InputView: () -> Other = { Other() } }\n" + "fun Scope.extensionRun() {\n" + " val value = InputView()\n" + " value.act()\n" + "}\n" + ), + }) + + callers = { + _find(result, ".run()", "caller"), + _find(result, "extensionRun()", "extension"), + } + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] in callers and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_companion_callable_shadows_constructor_inference(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "class InputView {\n" + " fun act() {}\n" + "}\n" + "class Other {\n" + " fun act() {}\n" + "}\n" + "class Caller {\n" + " companion object {\n" + " val InputView: () -> Other = { Other() }\n" + " }\n" + " fun run() {\n" + " val value = InputView()\n" + " value.act()\n" + " }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "caller") + acts = { + _find(result, ".act()", "inputview"), + _find(result, ".act()", "other"), + } + assert not any( + edge["source"] == run and edge["target"] in acts + for edge in _call_edges(result) + ) + + +def test_member_method_wins_over_same_named_unrelated_extension( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "InputView.kt": "class InputView { fun act() {} }\n", + "Calls.kt": ( + "class Other\n" + "fun Other.act() {}\n" + "fun run(view: InputView) { view.act() }\n" + ), + }) + + run = _find(result, "run()", "calls") + expected = _find(result, ".act()", "inputview") + wrong = _find(result, "act()", "calls") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_companion_extension_is_not_visible_as_top_level(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "class InputView { fun decorate() {} }\n" + "class Host {\n" + " companion object {\n" + " fun InputView.decorate() {}\n" + " }\n" + "}\n" + "fun run(value: InputView) { value.decorate() }\n" + ), + }) + + run = _find(result, "run()", "calls") + expected = _find(result, ".decorate()", "inputview") + wrong = _find(result, "decorate()", "calls") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_member_extension_arity_collision_is_left_unresolved( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "class InputView { fun act(value: Int) {} }\n" + "fun InputView.act() {}\n" + "fun run(view: InputView) { view.act() }\n" + ), + }) + + run = _find(result, "run()", "calls") + member = _find(result, ".act()", "inputview") + extension = _find(result, "act()", "calls") + assert not any( + edge["source"] == run and edge["target"] in {member, extension} + for edge in _call_edges(result) + ) + + +def test_extension_does_not_override_possible_inherited_member( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "open class Base { fun act() {} }\n" + "class Child : Base()\n" + "fun Child.act() {}\n" + "fun run(value: Child) { value.act() }\n" + ), + }) + + run = _find(result, "run()", "calls") + extension = _find(result, "act()", "calls") + assert not any( + edge["source"] == run and edge["target"] == extension + for edge in _call_edges(result) + ) + + +def test_member_extension_resolves_inside_its_dispatch_owner(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Extensions.kt": ( + "class InputView\n" + "class Host {\n" + " fun InputView.decorate() {}\n" + " fun run(view: InputView) { view.decorate() }\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "host") + decorate = _find(result, ".decorate()", "host") + assert any( + edge["source"] == run + and edge["target"] == decorate + and edge.get("confidence") == "EXTRACTED" + for edge in _call_edges(result) + ) + + +def test_inherited_member_and_member_extension_remain_resolvable( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "class InputView\n" + "interface Extensions {\n" + " fun InputView.decorate()\n" + "}\n" + "interface PrivateExtensions {\n" + " fun InputView.hidden()\n" + "}\n" + "open class Host {\n" + " fun inherited() {}\n" + " open fun InputView.decorate() {}\n" + " private fun InputView.hidden() {}\n" + "}\n" + "class Child : Host() {\n" + " override fun InputView.decorate() {}\n" + " fun run(view: InputView) {\n" + " inherited()\n" + " view.decorate()\n" + " }\n" + "}\n" + "class Sibling : Host(), Extensions {\n" + " fun siblingRun(view: InputView) { view.decorate() }\n" + "}\n" + "abstract class PrivateChild : Host(), PrivateExtensions {\n" + " fun privateRun(view: InputView) { view.hidden() }\n" + "}\n" + "class Outer {\n" + " private fun InputView.outerDecorate() {}\n" + " inner class Inner {\n" + " fun innerRun(view: InputView) { view.outerDecorate() }\n" + " }\n" + "}\n" + "fun typed(value: Child) { value.inherited() }\n" + ), + }) + + run = _find(result, ".run()", "child") + typed = _find(result, "typed()", "calls") + inherited = _find(result, ".inherited()", "host") + host_decorate = _find(result, ".decorate()", "host") + child_decorate = _find(result, ".decorate()", "child") + sibling_run = _find(result, ".siblingRun()", "sibling") + private_run = _find(result, ".privateRun()", "privatechild") + private_host = _find(result, ".hidden()", "host") + private_interface = _find(result, ".hidden()", "privateextensions") + inner_run = _find(result, ".innerRun()", "inner") + outer_decorate = _find(result, ".outerDecorate()", "outer") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, inherited) in calls + assert (typed, inherited) in calls + assert (run, child_decorate) in calls + assert (run, host_decorate) not in calls + assert (sibling_run, host_decorate) in calls + assert (private_run, private_interface) in calls + assert (private_run, private_host) not in calls + assert (inner_run, outer_decorate) in calls + + +def test_super_and_object_qualified_calls_keep_existing_resolution( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Calls.kt": ( + "interface Action {\n" + " fun act()\n" + "}\n" + "open class Base {\n" + " open fun act() {}\n" + "}\n" + "class Child : Base(), Action {\n" + " override fun act() {}\n" + " fun run() { super.act() }\n" + "}\n" + "class TypedChild : Base(), Action\n" + "object InputView {\n" + " fun show() {}\n" + "}\n" + "class Toolbar {\n" + " companion object {\n" + " fun render() {}\n" + " private fun secret() {}\n" + " }\n" + " class Nested {\n" + " fun nestedRun() { Toolbar.secret() }\n" + " }\n" + "}\n" + "fun open(value: TypedChild) {\n" + " value.act()\n" + " InputView.show()\n" + " Toolbar.render()\n" + "}\n" + ), + }) + + run = _find(result, ".run()", "child") + act = _find(result, ".act()", "base") + open_call = _find(result, "open()", "calls") + show = _find(result, ".show()", "inputview") + render = _find(result, "render()", "calls") + nested_run = _find(result, ".nestedRun()", "nested") + secret = _find(result, "secret()", "calls") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, act) in calls + assert (open_call, act) in calls + assert (open_call, show) in calls + assert (open_call, render) in calls + assert (nested_run, secret) in calls + + +def test_nested_receiver_type_resolves_with_enclosing_type_name( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Nested.kt": ( + "package a\n" + "class Outer {\n" + " class InputView { fun act() {} }\n" + "}\n" + "fun run(value: Outer.InputView) { value.act() }\n" + ), + }) + + run = _find(result, "run()", "nested") + act = _find(result, ".act()", "inputview") + assert any( + edge["source"] == run and edge["target"] == act + for edge in _call_edges(result) + ) + + +def test_imported_enclosing_type_resolves_nested_receiver(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "Default.kt": ( + "class Outer {\n" + " class InputView {\n" + " fun act() {}\n" + " }\n" + "}\n" + ), + "a/Outer.kt": ( + "package a\n" + "class Outer {\n" + " class InputView {\n" + " fun act() {}\n" + " }\n" + "}\n" + ), + "c/Caller.kt": ( + "package c\n" + "import a.Outer\n" + "fun run(value: Outer.InputView) { value.act() }\n" + ), + }) + + run = _find(result, "run()", "caller") + expected = _find(result, ".act()", "a_outer_inputview") + wrong = _find(result, ".act()", "default_inputview") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls + + +def test_aliased_extension_import_resolves_local_call_name(tmp_path: Path) -> None: + result = _extract(tmp_path, { + "a/Outer.kt": ( + "package a\n" + "class Outer {\n" + " class InputView\n" + "}\n" + ), + "b/Extensions.kt": ( + "package b\n" + "import a.Outer\n" + "fun Outer.InputView.decorate() {}\n" + ), + "c/Caller.kt": ( + "package c\n" + "import a.Outer\n" + "import b.decorate as adorn\n" + "fun run(view: Outer.InputView) { view.adorn() }\n" + ), + }) + + run = _find(result, "run()", "caller") + decorate = _find(result, "decorate()", "extensions") + assert any( + edge["source"] == run and edge["target"] == decorate + for edge in _call_edges(result) + ) + + +def test_kotlin_resolution_markers_do_not_leak_to_public_nodes( + tmp_path: Path, +) -> None: + result = _extract(tmp_path, { + "Extensions.kt": ( + "class InputView\n" + "fun InputView.decorate() {}\n" + "fun run(view: InputView) { view.decorate() }\n" + ), + }) + + assert all( + "_kotlin_extension_receiver_type" not in node + for node in result["nodes"] + ) + + +def test_package_resolution_survives_portable_ast_cache(tmp_path: Path) -> None: + shared_cache = tmp_path / "shared-cache" + + def extract_corpus(root: Path) -> dict: + files = { + "a/InputView.kt": "package a\nclass InputView { fun act() {} }\n", + "b/InputView.kt": "package b\nclass InputView { fun act() {} }\n", + "Caller.kt": ( + "class Caller { fun run(view: a.InputView) { view.act() } }\n" + ), + } + paths = [] + for name, body in files.items(): + path = root / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body, encoding="utf-8") + paths.append(path) + return extract(paths, cache_root=shared_cache, parallel=False) + + fresh = extract_corpus(tmp_path / "repo-a") + cached = extract_corpus(tmp_path / "repo-b") + + for result in (fresh, cached): + run = _find(result, ".run()", "caller") + expected = _find(result, ".act()", "a_inputview") + wrong = _find(result, ".act()", "b_inputview") + calls = {(edge["source"], edge["target"]) for edge in _call_edges(result)} + assert (run, expected) in calls + assert (run, wrong) not in calls