diff --git a/graphify/extract.py b/graphify/extract.py index 8380e0de..0cac5060 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -54,6 +54,82 @@ from graphify.extractors.zig import extract_zig # noqa: F401 from graphify.security import sanitize_metadata from graphify.paths import disambiguate_ambiguous_candidates +from graphify.extractors.models import LanguageConfig, _JS_CACHE_BYPASS_SUFFIXES, _NamespaceExportFact, _StarExportFact, _SymbolAliasFact, _SymbolDeclarationFact, _SymbolExportFact, _SymbolImportFact, _SymbolResolutionFacts, _SymbolUseFact, _WORKSPACE_PACKAGE_CACHE # noqa: E402,F401 + +from graphify.extractors.resolution import ( # noqa: E402,F401 + _DECLDEF_HEADER_SUFFIXES, + _DECLDEF_IMPL_SUFFIXES, + _EXPORT_CONDITION_PRIORITY, + _JS_INDEX_FILES, + _JS_PRIMITIVE_TYPES, + _JS_RESOLVE_EXTS, + _TSCONFIG_ALIAS_CACHE, + _VUE_SCRIPT_LANG_RE, + _VUE_SCRIPT_RE, + _WORKSPACE_MANIFEST_NAMES, + _apply_symbol_resolution_facts, + _augment_symbol_resolution_edges, + _collect_js_symbol_resolution_facts, + _collect_python_symbol_resolution_facts, + _contained_in_package, + _decldef_class_stem, + _disambiguate_colliding_node_ids, + _find_workspace_root, + _is_type_like_definition, + _js_call_identifier, + _js_default_export_name, + _js_default_import_name, + _js_export_clause, + _js_export_statement_is_star, + _js_exported_declaration_names, + _js_lexical_aliases, + _js_module_specifier, + _js_named_specifiers, + _js_namespace_export_name, + _js_source_path, + _js_top_level_function_bodies, + _load_tsconfig_aliases, + _load_workspace_packages, + _match_tsconfig_alias, + _merge_decl_def_classes, + _node_disambiguation_source_key, + _package_entry_candidates, + _parse_js_tree, + _parse_python_tree, + _pascal_class_stem_cache, + _pascal_project_root, + _pascal_resolve_class, + _pascal_resolve_unit, + _pascal_unit_cache, + _pnpm_workspace_globs, + _python_call_identifier, + _python_import_from_module, + _python_imported_names, + _python_top_level_function_bodies, + _read_tsconfig_aliases, + _resolve_c_include_path, + _resolve_cross_file_imports, + _resolve_cross_file_java_imports, + _resolve_export_target, + _resolve_java_type_references, + _resolve_js_import_path, + _resolve_js_import_target, + _resolve_js_module_path, + _resolve_lua_import_target, + _resolve_python_module_path, + _resolve_tsconfig_alias, + _resolve_workspace_import, + _source_key, + _strip_jsonc, + _ts_collect_type_refs, + _ts_heritage_clause_entries, + _ts_walk_class_members, + _vue_mask_non_script, + _walk_js_tree, + _walk_python_tree, + _workspace_globs, +) + _RECURSION_LIMIT = 10_000 # Language built-in globals that AST may classify as call targets when used as @@ -97,14 +173,6 @@ def _csharp_namespace_id(dotted_name: str) -> str: return f"csharp_namespace:{digest}" -_TSCONFIG_ALIAS_CACHE: dict[str, dict[str, list[str]]] = {} -_WORKSPACE_PACKAGE_CACHE: dict[str, dict[str, Path]] = {} -_WORKSPACE_MANIFEST_NAMES = ("pnpm-workspace.yaml", "package.json") -_JS_CACHE_BYPASS_SUFFIXES = {".js", ".jsx", ".mjs", ".ts", ".tsx", ".mts", ".cts", ".vue", ".svelte"} -_JS_RESOLVE_EXTS = (".ts", ".tsx", ".mts", ".cts", ".svelte", ".js", ".jsx", ".mjs") -_JS_INDEX_FILES = ("index.ts", "index.tsx", "index.svelte", "index.js", "index.jsx", "index.mjs") - - SEMANTIC_RELATIONS = frozenset({ "inherits", "implements", "mixes_in", "embeds", "references", "calls", "imports", "imports_from", "re_exports", "contains", "method", @@ -144,493 +212,13 @@ def _semantic_reference_edge( } -def _resolve_js_import_path(candidate: Path) -> Path: - """Resolve a JS/TS/Svelte import target to a local file when it exists.""" - candidate = Path(os.path.normpath(candidate)) - if candidate.is_file(): - return candidate - - # TS ESM convention: imports often spell .js/.jsx while source is .ts/.tsx. - if candidate.suffix == ".js": - ts_candidate = candidate.with_suffix(".ts") - if ts_candidate.is_file(): - return ts_candidate - elif candidate.suffix == ".jsx": - tsx_candidate = candidate.with_suffix(".tsx") - if tsx_candidate.is_file(): - return tsx_candidate - - # Append extensions to the full filename, which covers extensionless imports, - # multi-dot helpers, and Svelte 5 rune files like Foo.svelte.ts. - for ext in _JS_RESOLVE_EXTS: - with_ext = candidate.parent / f"{candidate.name}{ext}" - if with_ext.is_file(): - return with_ext - - # Only fall back to directory indexes after file candidates lose. - if candidate.is_dir(): - for index_name in _JS_INDEX_FILES: - index_candidate = candidate / index_name - if index_candidate.is_file(): - return index_candidate - - return candidate - - -def _strip_jsonc(text: str) -> str: - """Strip // line comments, /* */ block comments, and trailing commas from JSONC. - - Preserves string contents (including // and /* inside strings) by skipping over - quoted spans first. Required for tsconfig.json files generated by SvelteKit, - NestJS, Vite, T3, Astro, etc., which use JSONC by default (#700). - """ - # Remove block and line comments while leaving string literals untouched. - pattern = re.compile( - r'"(?:\\.|[^"\\])*"' # double-quoted string (with escapes) - r"|/\*.*?\*/" # /* block comment */ - r"|//[^\n]*", # // line comment - re.DOTALL, - ) - - def _replace(match: re.Match) -> str: - token = match.group(0) - if token.startswith('"'): - return token - return "" - - stripped = pattern.sub(_replace, text) - # Remove trailing commas before } or ] (allowing whitespace between). - stripped = re.sub(r",(\s*[}\]])", r"\1", stripped) - return stripped - - -def _read_tsconfig_aliases(tsconfig: Path, base_dir: Path, seen: set) -> dict[str, list[str]]: - """Recursively read path aliases from a tsconfig, following extends chains. - - Child config paths override parent. Circular extends are detected via seen set. - npm package configs (e.g. @tsconfig/svelte) are skipped since they're not on disk. - Handles JSONC (comments + trailing commas) which is the default tsconfig format - for SvelteKit, NestJS, Vite, T3, Astro, etc. (#700). - """ - if str(tsconfig) in seen: - return {} - seen.add(str(tsconfig)) - try: - raw = tsconfig.read_text(encoding="utf-8") - except Exception as e: - print(f" warning: could not read {tsconfig} ({type(e).__name__}: {e})", file=sys.stderr, flush=True) - return {} - try: - data = json.loads(raw) - except json.JSONDecodeError: - try: - data = json.loads(_strip_jsonc(raw)) - except json.JSONDecodeError as e: - print(f" warning: failed to parse {tsconfig} as JSON/JSONC ({e.msg} at line {e.lineno} col {e.colno})", file=sys.stderr, flush=True) - return {} - except Exception as e: - print(f" warning: failed to parse {tsconfig} ({type(e).__name__}: {e})", file=sys.stderr, flush=True) - return {} - - aliases: dict[str, list[str]] = {} - # `extends` may be a string or, since TypeScript 5.0, an array of paths. - # For an array, parents are processed in order with later entries - # overriding earlier ones; the extending config (paths below) overrides - # all parents. Without the list branch, an array `extends` raised - # `AttributeError: 'list' object has no attribute 'startswith'`, which - # _safe_extract turned into a skip of the whole file. - extends = data.get("extends") - if isinstance(extends, str): - extends_list = [extends] - elif isinstance(extends, list): - extends_list = [e for e in extends if isinstance(e, str)] - else: - extends_list = [] - for ext in extends_list: - # Skip scoped npm package configs (e.g. @tsconfig/svelte) — not on disk. - if not ext or ext.startswith("@"): - continue - extended_path = (base_dir / ext).resolve() - if not extended_path.suffix: - extended_path = extended_path.with_suffix(".json") - if extended_path.exists(): - aliases.update(_read_tsconfig_aliases(extended_path, extended_path.parent, seen)) - - # tsconfig `paths` are resolved relative to `baseUrl` (itself relative to - # the tsconfig's directory), not the tsconfig directory directly. Honoring - # baseUrl is required for the common monorepo / NestJS layout where - # baseUrl points at a subdirectory, e.g. baseUrl "./src" with - # "@services/*": ["services/*"] must resolve to /src/services rather - # than /services. Defaults to "." so configs without baseUrl (paths - # relative to the tsconfig dir, the TS 4.1+ behavior) keep working. - compiler_options = data.get("compilerOptions", {}) - base_url = compiler_options.get("baseUrl") or "." - paths_base = base_dir / base_url - paths = compiler_options.get("paths", {}) - for alias, targets in paths.items(): - if not targets: - continue - # Keep ALL targets in declared order — tsc tries each until one resolves - # on disk. Discarding the fallbacks (#1531) misresolved/dropped imports - # whose file lived at a non-first target. Preserve wildcard tokens in - # both sides until the resolver substitutes the captured segment, then - # normalizes the concrete path (#927). Empty/non-string entries are skipped. - target_patterns = [ - str(paths_base / t) - for t in targets - if isinstance(t, str) and t - ] - if target_patterns: - aliases[alias] = target_patterns - - return aliases - - -def _load_tsconfig_aliases(start_dir: Path) -> dict[str, list[str]]: - """Walk up from start_dir to find tsconfig.json and return compilerOptions.paths aliases. - - Follows extends chains so SvelteKit/Nuxt/NestJS inherited aliases are included. - Returns a dict mapping alias patterns to ordered resolved target patterns; - wildcard tokens remain intact for substitution during resolution (#927). - Result is cached by tsconfig path string. - """ - current = start_dir.resolve() - for candidate in [current, *current.parents]: - tsconfig = candidate / "tsconfig.json" - if tsconfig.exists(): - key = str(tsconfig) - if key not in _TSCONFIG_ALIAS_CACHE: - _TSCONFIG_ALIAS_CACHE[key] = _read_tsconfig_aliases(tsconfig, candidate, seen=set()) - return _TSCONFIG_ALIAS_CACHE[key] - return {} - - -def _match_tsconfig_alias(raw: str, pattern: str) -> "tuple[tuple[int, int], str, bool] | None": - """Return (specificity, captured text, is_wildcard) when pattern matches raw. - - Exact aliases win first. Wildcard aliases follow TypeScript's longest-prefix - rule. The final branch preserves Graphify's existing support for treating a - non-wildcard alias as a directory prefix, but only after real wildcard matches. - """ - if "*" in pattern: - if pattern.count("*") != 1: - return None - prefix, suffix = pattern.split("*", 1) - if not raw.startswith(prefix) or not raw.endswith(suffix): - return None - end = len(raw) - len(suffix) if suffix else len(raw) - if end < len(prefix): - return None - return (1, -len(prefix)), raw[len(prefix):end], True - - if raw == pattern: - return (0, -len(pattern)), "", False - - prefix = pattern.rstrip("/") - if prefix and raw.startswith(prefix + "/"): - return (2, -len(prefix)), raw[len(prefix):].lstrip("/"), False - return None - - -def _resolve_tsconfig_alias(raw: str, aliases: dict[str, list[str]]) -> "Path | None": - """Resolve `raw` against the most specific matching tsconfig alias pattern. - - Within that pattern, try targets in declared order and return the first whose - candidate resolves to a real file. If none exist, return the first candidate - so existing phantom/external-edge behavior stays unchanged. - """ - best: "tuple[tuple[int, int], str, bool, list[str]] | None" = None - for pattern, targets in aliases.items(): - match = _match_tsconfig_alias(raw, pattern) - if match is None: - continue - specificity, captured, is_wildcard = match - if best is None or specificity < best[0]: - best = specificity, captured, is_wildcard, targets - - if best is None: - return None - - _, captured, is_wildcard, targets = best - first = None - for target in targets: - if is_wildcard: - # TypeScript substitutes only when the matched star is non-empty. - substituted = target.replace("*", captured, 1) if captured else target - cand = Path(os.path.normpath(substituted)) - else: - cand = Path(target) - if captured: - cand = Path(os.path.normpath(cand / captured)) - resolved = _resolve_js_import_path(cand) - if resolved.is_file(): - return resolved - if first is None: - first = cand - return first - - -def _find_workspace_root(start_dir: Path) -> Path | None: - current = start_dir.resolve() - for candidate in [current, *current.parents]: - if (candidate / "pnpm-workspace.yaml").exists(): - return candidate - package_json = candidate / "package.json" - if package_json.is_file(): - try: - data = json.loads(package_json.read_text(encoding="utf-8")) - except Exception: - continue - if "workspaces" in data: - return candidate - return None - - -def _pnpm_workspace_globs(workspace_file: Path) -> list[str]: - globs: list[str] = [] - in_packages = False - for raw_line in workspace_file.read_text(encoding="utf-8", errors="replace").splitlines(): - line = raw_line.strip() - if not line or line.startswith("#"): - continue - if line.startswith("packages:"): - in_packages = True - continue - if in_packages and line.startswith("-"): - value = line[1:].strip().strip("'\"") - if value and not value.startswith("!"): - globs.append(value) - continue - if in_packages and not raw_line.startswith((" ", "\t")): - break - return globs - - -def _workspace_globs(root: Path) -> list[str]: - pnpm_workspace = root / "pnpm-workspace.yaml" - if pnpm_workspace.exists(): - return _pnpm_workspace_globs(pnpm_workspace) - - package_json = root / "package.json" - try: - data = json.loads(package_json.read_text(encoding="utf-8")) - except Exception: - return [] - - workspaces = data.get("workspaces") - if isinstance(workspaces, list): - return [item for item in workspaces if isinstance(item, str) and not item.startswith("!")] - if isinstance(workspaces, dict): - packages = workspaces.get("packages") - if isinstance(packages, list): - return [item for item in packages if isinstance(item, str) and not item.startswith("!")] - return [] - - -def _load_workspace_packages(start_dir: Path) -> dict[str, Path]: - root = _find_workspace_root(start_dir) - if root is None: - return {} - manifest_mtimes = tuple( - (name, (root / name).stat().st_mtime_ns) - for name in _WORKSPACE_MANIFEST_NAMES - if (root / name).is_file() - ) - key = str((root, manifest_mtimes)) - if key in _WORKSPACE_PACKAGE_CACHE: - return _WORKSPACE_PACKAGE_CACHE[key] - - packages: dict[str, Path] = {} - for pattern in _workspace_globs(root): - package_dirs: list[Path] = [root] if pattern in (".", "./") else list(root.glob(pattern)) - for package_dir in package_dirs: - manifest = package_dir / "package.json" - if not manifest.is_file(): - continue - try: - data = json.loads(manifest.read_text(encoding="utf-8")) - except Exception: - continue - name = data.get("name") - if isinstance(name, str) and name: - packages[name] = package_dir - _WORKSPACE_PACKAGE_CACHE[key] = packages - return packages - - # Condition keys consulted when resolving an `exports` target, in priority # order. `default` is Node's catch-all and must be consulted LAST so a more # specific condition (source/import/module/etc.) wins when several match. -_EXPORT_CONDITION_PRIORITY = ( - "source", "import", "module", "svelte", "types", "require", "default", -) - - -def _resolve_export_target(value: Any) -> str | None: - """Resolve an `exports` map value (string or condition object) to a - relative target string, honouring _EXPORT_CONDITION_PRIORITY for objects - and recursing into nested condition objects.""" - if isinstance(value, str): - return value - if isinstance(value, dict): - for cond in _EXPORT_CONDITION_PRIORITY: - v = value.get(cond) - if isinstance(v, str): - return v - if isinstance(v, dict): - nested = _resolve_export_target(v) - if nested: - return nested - return None - - -def _contained_in_package(resolved: Path, package_dir: Path) -> bool: - """Guard against `exports` targets that escape the package directory - (e.g. "./evil": "../../../etc/passwd"). Only accept paths that stay - within package_dir after resolution.""" - try: - return resolved.resolve().is_relative_to(package_dir.resolve()) - except ValueError: - return False - - -def _package_entry_candidates(package_dir: Path, subpath: str) -> list[Path]: - manifest = package_dir / "package.json" - manifest_data: dict[str, Any] = {} - try: - manifest_data = json.loads(manifest.read_text(encoding="utf-8")) - except Exception: - pass - - if subpath: - # Consult the package's `exports` subpath map before the bare-path - # fallback (#1308): "./browser" -> conditions -> file, plus single - # wildcard "./*" patterns. Targets that escape the package dir are - # rejected; resolution then falls through to the bare path. - exports = manifest_data.get("exports") - if isinstance(exports, dict): - subpath_key = "./" + subpath - target = _resolve_export_target(exports.get(subpath_key)) - if target: - candidate = package_dir / target - if _contained_in_package(candidate, package_dir): - return [candidate] - else: - for pattern, pattern_value in exports.items(): - if "*" in pattern and pattern.count("*") == 1: - prefix, suffix = pattern.split("*", 1) - if (subpath_key.startswith(prefix) - and (not suffix or subpath_key.endswith(suffix))): - matched = subpath_key[len(prefix):len(subpath_key) - len(suffix) if suffix else None] - resolved = _resolve_export_target(pattern_value) - if resolved and "*" in resolved: - candidate = package_dir / resolved.replace("*", matched) - if _contained_in_package(candidate, package_dir): - return [candidate] - return [package_dir / subpath] - - exports = manifest_data.get("exports") - if isinstance(exports, str): - return [package_dir / exports] - if isinstance(exports, dict): - dot_target = _resolve_export_target(exports.get(".")) - if dot_target: - return [package_dir / dot_target] - - candidates: list[Path] = [] - for key in ("svelte", "module", "main", "types"): - value = manifest_data.get(key) - if isinstance(value, str): - candidates.append(package_dir / value) - candidates.append(package_dir / "src/index") - candidates.append(package_dir / "index") - return candidates - - -def _resolve_workspace_import(raw: str, start_dir: Path) -> Path | None: - packages = _load_workspace_packages(start_dir) - for package_name, package_dir in packages.items(): - if raw == package_name: - subpath = "" - elif raw.startswith(package_name + "/"): - subpath = raw[len(package_name) + 1:] - else: - continue - for candidate in _package_entry_candidates(package_dir, subpath): - resolved = _resolve_js_import_path(candidate) - if resolved.is_file(): - return resolved - return None - - -def _resolve_js_module_path(raw: str | Path, start_dir: Path | None = None) -> Path | None: - """Resolve a JS/TS module path or specifier to a local source file. - - With a Path argument this preserves the path-based helper API used by - import-extension tests. With a string plus start_dir it resolves JS/TS - module specifiers including relative paths, tsconfig aliases, and workspace - packages. - """ - if isinstance(raw, Path): - return _resolve_js_import_path(raw) - if start_dir is None: - return _resolve_js_import_path(Path(raw)) - if raw.startswith("."): - return _resolve_js_import_path(start_dir / raw) - - aliases = _load_tsconfig_aliases(start_dir) - hit = _resolve_tsconfig_alias(raw, aliases) - if hit is not None: - return _resolve_js_import_path(hit) - - return _resolve_workspace_import(raw, start_dir) # ── LanguageConfig dataclass ───────────────────────────────────────────────── -@dataclass -class LanguageConfig: - ts_module: str # e.g. "tree_sitter_python" - ts_language_fn: str = "language" # attr to call: e.g. tslang.language() - - class_types: frozenset = frozenset() - function_types: frozenset = frozenset() - import_types: frozenset = frozenset() - call_types: frozenset = frozenset() - static_prop_types: frozenset = frozenset() - helper_fn_names: frozenset = frozenset() - container_bind_methods: frozenset = frozenset() - event_listener_properties: frozenset = frozenset() - - # Name extraction - name_field: str = "name" - name_fallback_child_types: tuple = () - - # Body detection - body_field: str = "body" - body_fallback_child_types: tuple = () # e.g. ("declaration_list", "compound_statement") - - # Call name extraction - call_function_field: str = "function" # field on call node for callee - call_accessor_node_types: frozenset = frozenset() # member/attribute nodes - call_accessor_field: str = "attribute" # field on accessor for method name - call_accessor_object_field: str = "" # field on accessor for the receiver/object - - # Stop recursion at these types in walk_calls - function_boundary_types: frozenset = frozenset() - - # Import handler: called for import nodes instead of generic handling - import_handler: Callable | None = None - - # Optional custom name resolver for functions (C, C++ declarator unwrapping) - resolve_function_name_fn: Callable | None = None - - # Extra label formatting for functions: if True, functions get "name()" label - function_label_parens: bool = True - - # Extra walk hook called after generic dispatch (for JS arrow functions, C# namespaces, etc.) - extra_walk_fn: Callable | None = None - # ── Generic helpers ─────────────────────────────────────────────────────────── @@ -1742,34 +1330,6 @@ def _import_python(node, source: bytes, file_nid: str, stem: str, edges: list, s }) -def _resolve_js_import_target(raw: str, str_path: str) -> "tuple[str, Path | None] | None": - """Resolve a JS/TS import path string to (target_nid, resolved_path). - - Handles relative paths, tsconfig path aliases, workspace packages, and - bare/scoped imports. - Returns None if `raw` is empty. - """ - if not raw: - return None - resolved_path = _resolve_js_module_path(raw, Path(str_path).parent) - if resolved_path is not None: - return _make_id(str(resolved_path)), resolved_path - module_name = raw.split("/")[-1] - if not module_name: - return None - # Unresolved: relative/absolute, tsconfig-alias and workspace resolution have - # all run and failed, so this is an external package (or a dangling local - # path). Namespace the id with the "ref" prefix — the J-4 convention already - # used for tsconfig `extends`/`$ref` externals — so it can NEVER collapse to - # the same _make_id as a local file/symbol node. Without it, the bare - # last-segment id (e.g. "tailwindcss/colors" -> "colors") collides with any - # unrelated local file of that stem via build.py's pre-migration alias index, - # producing a confident (EXTRACTED) cross-language phantom imports_from edge - # (#1638). The ref-namespaced target has no node, so build drops it as an - # external reference — the correct outcome for a third-party import. - return _make_id("ref", raw), None - - def _import_js(node, source: bytes, file_nid: str, stem: str, edges: list, str_path: str, scope_stack: list[str] | None = None) -> None: is_reexport = node.type == "export_statement" # Only handle export_statement if it has a `from` clause (re-export). @@ -1973,20 +1533,6 @@ def _import_java(node, source: bytes, file_nid: str, stem: str, edges: list, str break -def _resolve_c_include_path(raw: str, str_path: str) -> "Path | None": - """Resolve a quoted #include path to a real file on disk. - - Searches relative to the including file's directory. Returns None for - system headers (<...>) or paths that don't exist on disk. - """ - if not raw: - return None - candidate = (Path(str_path).parent / raw).resolve() - if candidate.is_file(): - return candidate - return None - - def _import_c(node, source: bytes, file_nid: str, stem: str, edges: list, str_path: str, scope_stack: list[str] | None = None) -> None: for child in node.children: if child.type in ("string_literal", "system_lib_string", "string"): @@ -3101,44 +2647,6 @@ _PHP_CONFIG = LanguageConfig( ) -def _resolve_lua_import_target(raw_module: str, str_path: str) -> str: - """Resolve a Lua require() module name to a node id. - - Lua module names use dots as path separators: `require("pkg.b")` looks for - `pkg/b.lua` (or `pkg/b/init.lua`) relative to a package root. We probe the - importing file's directory and walk upward looking for a matching file on - disk; if found, the returned id matches the file node id `_extract_generic` - assigns to that file (`_make_id(str(path))`), so the edge lands on a real - node. When nothing matches, fall back to `_make_id` of the full dotted - module name so cross-file resolution can still complete via the symbol - resolution pass instead of dropping the edge entirely (#1075). - """ - if not raw_module: - return "" - rel = raw_module.replace(".", "/") - try: - start_dir = Path(str_path).parent - except Exception: - start_dir = None - if start_dir is not None: - probe = start_dir - # Walk up a few levels so requires from nested files still resolve when - # the package root is above the importing file. - for _ in range(6): - for suffix in (".lua", ".luau"): - cand = probe / f"{rel}{suffix}" - if cand.is_file(): - return _make_id(str(cand)) - for suffix in (".lua", ".luau"): - cand = probe / rel / f"init{suffix}" - if cand.is_file(): - return _make_id(str(cand)) - if probe.parent == probe: - break - probe = probe.parent - return _make_id(raw_module) - - def _import_lua(node, source: bytes, file_nid: str, stem: str, edges: list, str_path: str, scope_stack: list[str] | None = None) -> None: """Extract require('module') from Lua variable_declaration nodes.""" text = _read_text(node, source) @@ -6143,40 +5651,6 @@ def extract_astro(path: Path) -> dict: # The open-tag matcher skips over quoted attribute values so a `>` inside one # (e.g. Vue 3.3+ generic components: ` close tag - pos = m.end() - if lang is None: - lang_m = _VUE_SCRIPT_LANG_RE.search(m.group(1)) - if lang_m: - lang = lang_m.group(1).lower() - out.append(_blank(src[pos:])) - return "".join(out), lang def extract_vue(path: Path) -> dict: @@ -6716,145 +6190,6 @@ def extract_julia(path: Path) -> dict: # ── Cross-file import resolution ────────────────────────────────────────────── -def _source_key(source_file: str, root: Path) -> str: - if not source_file: - return "" - source_path = Path(source_file) - try: - return str(source_path.resolve().relative_to(root)) - except Exception: - return str(source_path) - - -def _node_disambiguation_source_key(node: dict, root: Path) -> str: - source_file = str(node.get("source_file", "")) - if source_file: - return _source_key(source_file, root) - return _source_key(str(node.get("origin_file", "")), root) - - -def _disambiguate_colliding_node_ids( - nodes: list[dict], - edges: list[dict], - raw_calls: list[dict], - root: Path, -) -> None: - """Rewrite only colliding node IDs, using source path as the disambiguator. - - Module anchor nodes (#1327) are exempt: ``import CoreKit`` from three files - yields three ``type=module`` nodes with the same id but different - source_files. Those are the *same* module, not distinct same-named symbols, - so they must collapse to one shared node — disambiguating them by path would - scatter a single module across N file-qualified duplicates. - """ - by_id: dict[str, list[dict]] = {} - for node in nodes: - if node.get("type") in ("module", "namespace"): - continue - nid = node.get("id") - if isinstance(nid, str) and nid: - by_id.setdefault(nid, []).append(node) - - remap: dict[tuple[str, str], str] = {} - ambiguous_ids: set[str] = set() - for old_id, group in by_id.items(): - source_keys = {_node_disambiguation_source_key(node, root) for node in group} - if len(group) < 2 or len(source_keys) < 2: - continue - ambiguous_ids.add(old_id) - # Salt the colliding id with the *path* it came from. The naive salt is - # ``_make_id(source_key, old_id)`` — source_key is the raw repo-relative - # path. But _make_id collapses every separator, so two DISTINCT paths - # whose only difference is a separator-vs-inner-punctuation swap - # (``a/b/c.md`` vs ``a.b/c.md``, ``foo/bar_baz.md`` vs ``foo_bar/baz.md``) - # normalize to the SAME salted id and still collide (#1522 — the residual - # of #1504 the 0.9.0 full-path stem didn't reach). When that happens, - # append a short stable hash of the *raw* source_key, which IS injective - # over distinct paths, so the colliders separate. Computed in code from - # source_file (never trusted from the LLM), so AST↔semantic parity holds. - naive: dict[str, str] = {} # source_key -> _make_id(source_key, old_id) - for source_key in source_keys: - if source_key: - naive[source_key] = _make_id(source_key, old_id) - # source_keys that, after normalization, are not unique among themselves. - seen: dict[str, int] = {} - for nid in naive.values(): - seen[nid] = seen.get(nid, 0) + 1 - needs_hash = {sk for sk, nid in naive.items() if seen.get(nid, 0) > 1} - for node in group: - source_key = _node_disambiguation_source_key(node, root) - if not source_key: - continue - if source_key in needs_hash: - salt = hashlib.sha1(source_key.encode("utf-8")).hexdigest()[:6] - new_id = _make_id(source_key, old_id, salt) - else: - new_id = naive.get(source_key) or _make_id(source_key, old_id) - remap[(old_id, source_key)] = new_id - if new_id != old_id: - node["id"] = new_id - - if not remap: - return - - unambiguous_remaps: dict[str, str] = {} - for old_id, group in by_id.items(): - if old_id in ambiguous_ids: - continue - candidates = { - node["id"] for node in group - if isinstance(node.get("id"), str) and node["id"] != old_id - } - if len(candidates) == 1: - unambiguous_remaps[old_id] = next(iter(candidates)) - - # A C/ObjC/C++ `#include "foo.h"` / `#import "foo.h"` resolves to the header's - # file node, but `foo.h` and its sibling `foo.c`/`foo.m`/`foo.cpp` collapse to - # the same `foo` file id, so disambiguation salts them apart by path. A - # cross-file import edge from a THIRD file carries neither salt's source_key, so - # the (target, edge_source_key) lookup misses and the edge dangles on the now - # dead `foo` id. Repoint those import edges to the HEADER variant (the include - # always targeted the header), keyed by the original colliding id (#1475). - _HEADER_SUFFIXES = (".h", ".hpp", ".hh", ".hxx") - header_remaps: dict[str, str] = {} - for old_id in ambiguous_ids: - for node in by_id.get(old_id, []): - sk = _node_disambiguation_source_key(node, root) - if sk and Path(sk).suffix.lower() in _HEADER_SUFFIXES: - new_id = remap.get((old_id, sk)) - if new_id: - header_remaps[old_id] = new_id - break - - for edge in edges: - edge_source_key = _source_key(str(edge.get("source_file", "")), root) - source_key = (edge.get("source", ""), edge_source_key) - target_key = (edge.get("target", ""), edge_source_key) - if source_key in remap: - edge["source"] = remap[source_key] - elif edge.get("source") in unambiguous_remaps: - edge["source"] = unambiguous_remaps[str(edge["source"])] - # imports/imports_from always target a header file, so they must resolve to - # the header variant BEFORE the same-source-file salt is considered. Keying - # the import target by the importer's own source file mis-points a `.m` - # importing its own `.h` back at itself (self-loop), and is wrong for any - # cross-file import whose importer shares the colliding id (#1475). - if (edge.get("relation") in ("imports", "imports_from") - and edge.get("target") in header_remaps): - edge["target"] = header_remaps[str(edge["target"])] - elif target_key in remap: - edge["target"] = remap[target_key] - elif edge.get("target") in unambiguous_remaps: - edge["target"] = unambiguous_remaps[str(edge["target"])] - - for raw_call in raw_calls: - call_source_key = _source_key(str(raw_call.get("source_file", "")), root) - caller_key = (raw_call.get("caller_nid", ""), call_source_key) - if caller_key in remap: - raw_call["caller_nid"] = remap[caller_key] - elif raw_call.get("caller_nid") in unambiguous_remaps: - raw_call["caller_nid"] = unambiguous_remaps[str(raw_call["caller_nid"])] - def _canonicalize_csharp_namespace_nodes(all_nodes: list[dict], all_edges: list[dict]) -> None: """Collapse duplicate C# namespace node entries to one canonical node per label.""" @@ -6968,19 +6303,6 @@ def _node_label_key(node: dict, fold: bool = False) -> str: return key.lower() if fold else key -def _is_type_like_definition(node: dict) -> bool: - if node.get("type") == "namespace": - return False - label = str(node.get("label", "")).strip() - if not label: - return False - if label.endswith(")") or label.startswith("."): - return False - if "." in label: - return False - return node.get("file_type") == "code" - - def _rewire_unique_stub_nodes(nodes: list[dict], edges: list[dict]) -> None: """Map unresolved no-source stubs to a unique real definition with the same label.""" real_by_label: dict[str, list[dict]] = {} # exact-case (all languages) @@ -7052,555 +6374,6 @@ def _rewire_unique_stub_nodes(nodes: list[dict], edges: list[dict]) -> None: nodes[:] = [node for node in nodes if node.get("id") not in drop_ids] -def _js_source_path(source_file: str, root: Path) -> Path | None: - if not source_file: - return None - path = Path(source_file) - if not path.is_absolute(): - path = root / path - try: - return path.resolve() - except Exception: - return path - - -@dataclass(frozen=True) -class _SymbolDeclarationFact: - file_path: Path - name: str - line: int - - -@dataclass(frozen=True) -class _SymbolImportFact: - file_path: Path - local_name: str - target_path: Path - imported_name: str - line: int - - -@dataclass(frozen=True) -class _SymbolAliasFact: - file_path: Path - alias: str - target_name: str - line: int - - -@dataclass(frozen=True) -class _SymbolExportFact: - file_path: Path - exported_name: str - line: int - local_name: str | None = None - target_path: Path | None = None - target_name: str | None = None - - -@dataclass(frozen=True) -class _StarExportFact: - file_path: Path - target_path: Path - line: int - - -@dataclass(frozen=True) -class _NamespaceExportFact: - file_path: Path - exported_name: str - target_path: Path - line: int - - -@dataclass(frozen=True) -class _SymbolUseFact: - file_path: Path - source_id: str - local_name: str - relation: str - context: str - line: int - - -@dataclass -class _SymbolResolutionFacts: - declarations: list[_SymbolDeclarationFact] = field(default_factory=list) - imports: list[_SymbolImportFact] = field(default_factory=list) - aliases: list[_SymbolAliasFact] = field(default_factory=list) - exports: list[_SymbolExportFact] = field(default_factory=list) - star_exports: list[_StarExportFact] = field(default_factory=list) - namespace_exports: list[_NamespaceExportFact] = field(default_factory=list) - uses: list[_SymbolUseFact] = field(default_factory=list) - # File-to-file submodule imports from `from pkg import submod` (#1146). - # Each entry is (importing_file, submodule_file, line). - module_imports: list[tuple[Path, Path, int]] = field(default_factory=list) - - -def _apply_symbol_resolution_facts( - paths: list[Path], - nodes: list[dict], - edges: list[dict], - root: Path, - facts: _SymbolResolutionFacts, -) -> None: - """Apply language-provided import/export/use facts to graph edges.""" - if not ( - facts.declarations - or facts.imports - or facts.aliases - or facts.exports - or facts.star_exports - or facts.namespace_exports - or facts.uses - or facts.module_imports - ): - return - - path_by_resolved = {path.resolve(): path for path in paths} - source_file_id = {path.resolve(): _make_id(str(path)) for path in paths} - symbol_nodes: dict[tuple[Path, str], str] = {} - for node in nodes: - source_path = _js_source_path(str(node.get("source_file", "")), root) - if source_path is None: - continue - label = str(node.get("label", "")).strip().strip("()").lstrip(".") - if label and node.get("id"): - symbol_nodes[(source_path, label)] = str(node["id"]) - - def ensure_symbol_node(path: Path, name: str, line: int) -> str: - resolved_path = path.resolve() - existing = symbol_nodes.get((resolved_path, name)) - if existing is not None: - return existing - node_id = _make_id(_file_stem(path), name) - symbol_nodes[(resolved_path, name)] = node_id - nodes.append({ - "id": node_id, - "label": name, - "file_type": "code", - "source_file": str(path), - "source_location": f"L{line}", - }) - return node_id - - existing_edges = { - ( - str(edge.get("source")), - str(edge.get("target")), - str(edge.get("relation")), - str(edge.get("context") or ""), - ) - for edge in edges - } - - def add_edge(source: str, target: str, relation: str, context: str, line: int, source_path: Path) -> None: - key = (source, target, relation, context or "") - if key in existing_edges: - return - existing_edges.add(key) - edges.append({ - "source": source, - "target": target, - "relation": relation, - "context": context, - "confidence": "EXTRACTED", - "source_file": str(source_path), - "source_location": f"L{line}", - "weight": 1.0, - }) - - for declaration in facts.declarations: - ensure_symbol_node(declaration.file_path, declaration.name, declaration.line) - - local_aliases_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} - for import_fact in facts.imports: - file_path = import_fact.file_path.resolve() - local_aliases_by_file.setdefault(file_path, {})[import_fact.local_name] = ( - import_fact.target_path.resolve(), - import_fact.imported_name, - ) - - pending_aliases_by_file: dict[Path, list[_SymbolAliasFact]] = {} - for alias_fact in facts.aliases: - pending_aliases_by_file.setdefault(alias_fact.file_path.resolve(), []).append(alias_fact) - - for file_path, aliases in pending_aliases_by_file.items(): - local_aliases = local_aliases_by_file.setdefault(file_path, {}) - changed = True - while changed: - changed = False - for alias_fact in aliases: - if alias_fact.alias in local_aliases: - continue - origin = local_aliases.get(alias_fact.target_name) - if origin is not None: - local_aliases[alias_fact.alias] = origin - changed = True - - named_exports_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} - star_exports_by_file: dict[Path, list[Path]] = {} - - for star_fact in facts.star_exports: - source_path = star_fact.file_path.resolve() - target_path = star_fact.target_path.resolve() - star_exports_by_file.setdefault(source_path, []).append(target_path) - source_id = source_file_id.get(source_path) - if source_id is not None: - add_edge( - source_id, - _make_id(str(path_by_resolved.get(target_path, target_path))), - "re_exports", - "export", - star_fact.line, - star_fact.file_path, - ) - - for namespace_fact in facts.namespace_exports: - source_path = namespace_fact.file_path.resolve() - target_path = namespace_fact.target_path.resolve() - namespace_id = ensure_symbol_node( - namespace_fact.file_path, - namespace_fact.exported_name, - namespace_fact.line, - ) - named_exports_by_file.setdefault(source_path, {})[ - namespace_fact.exported_name - ] = (source_path, namespace_fact.exported_name) - source_id = source_file_id.get(source_path) - if source_id is not None: - add_edge( - source_id, - namespace_id, - "contains", - "namespace_export", - namespace_fact.line, - namespace_fact.file_path, - ) - add_edge( - source_id, - _make_id(str(path_by_resolved.get(target_path, target_path))), - "re_exports", - "export", - namespace_fact.line, - namespace_fact.file_path, - ) - - for export_fact in facts.exports: - file_path = export_fact.file_path.resolve() - origin: tuple[Path, str] | None = None - if export_fact.target_path is not None and export_fact.target_name is not None: - origin = (export_fact.target_path.resolve(), export_fact.target_name) - elif export_fact.local_name is not None: - origin = local_aliases_by_file.get(file_path, {}).get(export_fact.local_name) - if origin is None and (file_path, export_fact.local_name) in symbol_nodes: - origin = (file_path, export_fact.local_name) - if origin is None: - continue - named_exports_by_file.setdefault(file_path, {})[export_fact.exported_name] = origin - if origin[0] != file_path: - source_id = source_file_id.get(file_path) - if source_id is not None: - add_edge( - source_id, - _make_id(str(path_by_resolved.get(origin[0], origin[0]))), - "re_exports", - "export", - export_fact.line, - export_fact.file_path, - ) - - def resolve_exported_origin(target_path: Path, imported_name: str, seen: set[tuple[Path, str]] | None = None) -> tuple[Path, str]: - target_path = target_path.resolve() - key = (target_path, imported_name) - if seen is None: - seen = set() - if key in seen: - return key - seen.add(key) - origin = named_exports_by_file.get(target_path, {}).get(imported_name) - if origin is not None: - return resolve_exported_origin(origin[0], origin[1], seen) - for star_target in star_exports_by_file.get(target_path, []): - star_key = (star_target, imported_name) - if star_key in symbol_nodes: - return star_key - resolved = resolve_exported_origin(star_target, imported_name, seen) - if resolved in symbol_nodes: - return resolved - return key - - for import_fact in facts.imports: - source_id = source_file_id.get(import_fact.file_path.resolve()) - if source_id is None: - continue - origin_path, origin_symbol = resolve_exported_origin( - import_fact.target_path, - import_fact.imported_name, - ) - target_id = symbol_nodes.get((origin_path, origin_symbol)) - if target_id is None: - continue - add_edge( - source_id, - target_id, - "imports", - "import", - import_fact.line, - import_fact.file_path, - ) - - # #1146: emit file-to-file imports_from edges for package-form submodule imports. - for from_path, to_path, line in facts.module_imports: - try: - from_rel = from_path.relative_to(root) - to_rel = to_path.relative_to(root) - except ValueError: - continue - source_id = _make_id(_file_stem(from_rel)) - target_id = _make_id(_file_stem(to_rel)) - add_edge(source_id, target_id, "imports_from", "submodule_import", line, from_path) - - for use_fact in facts.uses: - file_path = use_fact.file_path.resolve() - target_id = None - unresolved_origin = local_aliases_by_file.get(file_path, {}).get(use_fact.local_name) - if unresolved_origin is not None: - origin_path, origin_symbol = resolve_exported_origin(*unresolved_origin) - target_id = symbol_nodes.get((origin_path, origin_symbol)) - if target_id is None and use_fact.relation in ("inherits", "implements"): - # Same-file fallback for HERITAGE only: a base declared in the same - # file (`class X extends Y`, `interface A extends B`) has no import - # alias, so resolve it directly against the file's own symbol nodes. - # Scoped to heritage because same-file calls/uses already resolve via - # the dedicated call-graph pass; widening this would duplicate those - # edges. Import resolution still takes precedence (#1095). - target_id = symbol_nodes.get((file_path, use_fact.local_name)) - if target_id is None: - continue - add_edge( - use_fact.source_id, - target_id, - use_fact.relation, - use_fact.context, - use_fact.line, - use_fact.file_path, - ) - - -def _parse_js_tree(path: Path): - try: - from tree_sitter import Language, Parser - # .vue embeds the script in non-JS markup; mask it out and parse the - # close tag + pos = m.end() + if lang is None: + lang_m = _VUE_SCRIPT_LANG_RE.search(m.group(1)) + if lang_m: + lang = lang_m.group(1).lower() + out.append(_blank(src[pos:])) + return "".join(out), lang + +def _source_key(source_file: str, root: Path) -> str: + if not source_file: + return "" + source_path = Path(source_file) + try: + return str(source_path.resolve().relative_to(root)) + except Exception: + return str(source_path) + +def _node_disambiguation_source_key(node: dict, root: Path) -> str: + source_file = str(node.get("source_file", "")) + if source_file: + return _source_key(source_file, root) + return _source_key(str(node.get("origin_file", "")), root) + +def _disambiguate_colliding_node_ids( + nodes: list[dict], + edges: list[dict], + raw_calls: list[dict], + root: Path, +) -> None: + """Rewrite only colliding node IDs, using source path as the disambiguator. + + Module anchor nodes (#1327) are exempt: ``import CoreKit`` from three files + yields three ``type=module`` nodes with the same id but different + source_files. Those are the *same* module, not distinct same-named symbols, + so they must collapse to one shared node — disambiguating them by path would + scatter a single module across N file-qualified duplicates. + """ + by_id: dict[str, list[dict]] = {} + for node in nodes: + if node.get("type") in ("module", "namespace"): + continue + nid = node.get("id") + if isinstance(nid, str) and nid: + by_id.setdefault(nid, []).append(node) + + remap: dict[tuple[str, str], str] = {} + ambiguous_ids: set[str] = set() + for old_id, group in by_id.items(): + source_keys = {_node_disambiguation_source_key(node, root) for node in group} + if len(group) < 2 or len(source_keys) < 2: + continue + ambiguous_ids.add(old_id) + # Salt the colliding id with the *path* it came from. The naive salt is + # ``_make_id(source_key, old_id)`` — source_key is the raw repo-relative + # path. But _make_id collapses every separator, so two DISTINCT paths + # whose only difference is a separator-vs-inner-punctuation swap + # (``a/b/c.md`` vs ``a.b/c.md``, ``foo/bar_baz.md`` vs ``foo_bar/baz.md``) + # normalize to the SAME salted id and still collide (#1522 — the residual + # of #1504 the 0.9.0 full-path stem didn't reach). When that happens, + # append a short stable hash of the *raw* source_key, which IS injective + # over distinct paths, so the colliders separate. Computed in code from + # source_file (never trusted from the LLM), so AST↔semantic parity holds. + naive: dict[str, str] = {} # source_key -> _make_id(source_key, old_id) + for source_key in source_keys: + if source_key: + naive[source_key] = _make_id(source_key, old_id) + # source_keys that, after normalization, are not unique among themselves. + seen: dict[str, int] = {} + for nid in naive.values(): + seen[nid] = seen.get(nid, 0) + 1 + needs_hash = {sk for sk, nid in naive.items() if seen.get(nid, 0) > 1} + for node in group: + source_key = _node_disambiguation_source_key(node, root) + if not source_key: + continue + if source_key in needs_hash: + salt = hashlib.sha1(source_key.encode("utf-8")).hexdigest()[:6] + new_id = _make_id(source_key, old_id, salt) + else: + new_id = naive.get(source_key) or _make_id(source_key, old_id) + remap[(old_id, source_key)] = new_id + if new_id != old_id: + node["id"] = new_id + + if not remap: + return + + unambiguous_remaps: dict[str, str] = {} + for old_id, group in by_id.items(): + if old_id in ambiguous_ids: + continue + candidates = { + node["id"] for node in group + if isinstance(node.get("id"), str) and node["id"] != old_id + } + if len(candidates) == 1: + unambiguous_remaps[old_id] = next(iter(candidates)) + + # A C/ObjC/C++ `#include "foo.h"` / `#import "foo.h"` resolves to the header's + # file node, but `foo.h` and its sibling `foo.c`/`foo.m`/`foo.cpp` collapse to + # the same `foo` file id, so disambiguation salts them apart by path. A + # cross-file import edge from a THIRD file carries neither salt's source_key, so + # the (target, edge_source_key) lookup misses and the edge dangles on the now + # dead `foo` id. Repoint those import edges to the HEADER variant (the include + # always targeted the header), keyed by the original colliding id (#1475). + _HEADER_SUFFIXES = (".h", ".hpp", ".hh", ".hxx") + header_remaps: dict[str, str] = {} + for old_id in ambiguous_ids: + for node in by_id.get(old_id, []): + sk = _node_disambiguation_source_key(node, root) + if sk and Path(sk).suffix.lower() in _HEADER_SUFFIXES: + new_id = remap.get((old_id, sk)) + if new_id: + header_remaps[old_id] = new_id + break + + for edge in edges: + edge_source_key = _source_key(str(edge.get("source_file", "")), root) + source_key = (edge.get("source", ""), edge_source_key) + target_key = (edge.get("target", ""), edge_source_key) + if source_key in remap: + edge["source"] = remap[source_key] + elif edge.get("source") in unambiguous_remaps: + edge["source"] = unambiguous_remaps[str(edge["source"])] + # imports/imports_from always target a header file, so they must resolve to + # the header variant BEFORE the same-source-file salt is considered. Keying + # the import target by the importer's own source file mis-points a `.m` + # importing its own `.h` back at itself (self-loop), and is wrong for any + # cross-file import whose importer shares the colliding id (#1475). + if (edge.get("relation") in ("imports", "imports_from") + and edge.get("target") in header_remaps): + edge["target"] = header_remaps[str(edge["target"])] + elif target_key in remap: + edge["target"] = remap[target_key] + elif edge.get("target") in unambiguous_remaps: + edge["target"] = unambiguous_remaps[str(edge["target"])] + + for raw_call in raw_calls: + call_source_key = _source_key(str(raw_call.get("source_file", "")), root) + caller_key = (raw_call.get("caller_nid", ""), call_source_key) + if caller_key in remap: + raw_call["caller_nid"] = remap[caller_key] + elif raw_call.get("caller_nid") in unambiguous_remaps: + raw_call["caller_nid"] = unambiguous_remaps[str(raw_call["caller_nid"])] + +def _is_type_like_definition(node: dict) -> bool: + if node.get("type") == "namespace": + return False + label = str(node.get("label", "")).strip() + if not label: + return False + if label.endswith(")") or label.startswith("."): + return False + if "." in label: + return False + return node.get("file_type") == "code" + +def _js_source_path(source_file: str, root: Path) -> Path | None: + if not source_file: + return None + path = Path(source_file) + if not path.is_absolute(): + path = root / path + try: + return path.resolve() + except Exception: + return path + +def _apply_symbol_resolution_facts( + paths: list[Path], + nodes: list[dict], + edges: list[dict], + root: Path, + facts: _SymbolResolutionFacts, +) -> None: + """Apply language-provided import/export/use facts to graph edges.""" + if not ( + facts.declarations + or facts.imports + or facts.aliases + or facts.exports + or facts.star_exports + or facts.namespace_exports + or facts.uses + or facts.module_imports + ): + return + + path_by_resolved = {path.resolve(): path for path in paths} + source_file_id = {path.resolve(): _make_id(str(path)) for path in paths} + symbol_nodes: dict[tuple[Path, str], str] = {} + for node in nodes: + source_path = _js_source_path(str(node.get("source_file", "")), root) + if source_path is None: + continue + label = str(node.get("label", "")).strip().strip("()").lstrip(".") + if label and node.get("id"): + symbol_nodes[(source_path, label)] = str(node["id"]) + + def ensure_symbol_node(path: Path, name: str, line: int) -> str: + resolved_path = path.resolve() + existing = symbol_nodes.get((resolved_path, name)) + if existing is not None: + return existing + node_id = _make_id(_file_stem(path), name) + symbol_nodes[(resolved_path, name)] = node_id + nodes.append({ + "id": node_id, + "label": name, + "file_type": "code", + "source_file": str(path), + "source_location": f"L{line}", + }) + return node_id + + existing_edges = { + ( + str(edge.get("source")), + str(edge.get("target")), + str(edge.get("relation")), + str(edge.get("context") or ""), + ) + for edge in edges + } + + def add_edge(source: str, target: str, relation: str, context: str, line: int, source_path: Path) -> None: + key = (source, target, relation, context or "") + if key in existing_edges: + return + existing_edges.add(key) + edges.append({ + "source": source, + "target": target, + "relation": relation, + "context": context, + "confidence": "EXTRACTED", + "source_file": str(source_path), + "source_location": f"L{line}", + "weight": 1.0, + }) + + for declaration in facts.declarations: + ensure_symbol_node(declaration.file_path, declaration.name, declaration.line) + + local_aliases_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} + for import_fact in facts.imports: + file_path = import_fact.file_path.resolve() + local_aliases_by_file.setdefault(file_path, {})[import_fact.local_name] = ( + import_fact.target_path.resolve(), + import_fact.imported_name, + ) + + pending_aliases_by_file: dict[Path, list[_SymbolAliasFact]] = {} + for alias_fact in facts.aliases: + pending_aliases_by_file.setdefault(alias_fact.file_path.resolve(), []).append(alias_fact) + + for file_path, aliases in pending_aliases_by_file.items(): + local_aliases = local_aliases_by_file.setdefault(file_path, {}) + changed = True + while changed: + changed = False + for alias_fact in aliases: + if alias_fact.alias in local_aliases: + continue + origin = local_aliases.get(alias_fact.target_name) + if origin is not None: + local_aliases[alias_fact.alias] = origin + changed = True + + named_exports_by_file: dict[Path, dict[str, tuple[Path, str]]] = {} + star_exports_by_file: dict[Path, list[Path]] = {} + + for star_fact in facts.star_exports: + source_path = star_fact.file_path.resolve() + target_path = star_fact.target_path.resolve() + star_exports_by_file.setdefault(source_path, []).append(target_path) + source_id = source_file_id.get(source_path) + if source_id is not None: + add_edge( + source_id, + _make_id(str(path_by_resolved.get(target_path, target_path))), + "re_exports", + "export", + star_fact.line, + star_fact.file_path, + ) + + for namespace_fact in facts.namespace_exports: + source_path = namespace_fact.file_path.resolve() + target_path = namespace_fact.target_path.resolve() + namespace_id = ensure_symbol_node( + namespace_fact.file_path, + namespace_fact.exported_name, + namespace_fact.line, + ) + named_exports_by_file.setdefault(source_path, {})[ + namespace_fact.exported_name + ] = (source_path, namespace_fact.exported_name) + source_id = source_file_id.get(source_path) + if source_id is not None: + add_edge( + source_id, + namespace_id, + "contains", + "namespace_export", + namespace_fact.line, + namespace_fact.file_path, + ) + add_edge( + source_id, + _make_id(str(path_by_resolved.get(target_path, target_path))), + "re_exports", + "export", + namespace_fact.line, + namespace_fact.file_path, + ) + + for export_fact in facts.exports: + file_path = export_fact.file_path.resolve() + origin: tuple[Path, str] | None = None + if export_fact.target_path is not None and export_fact.target_name is not None: + origin = (export_fact.target_path.resolve(), export_fact.target_name) + elif export_fact.local_name is not None: + origin = local_aliases_by_file.get(file_path, {}).get(export_fact.local_name) + if origin is None and (file_path, export_fact.local_name) in symbol_nodes: + origin = (file_path, export_fact.local_name) + if origin is None: + continue + named_exports_by_file.setdefault(file_path, {})[export_fact.exported_name] = origin + if origin[0] != file_path: + source_id = source_file_id.get(file_path) + if source_id is not None: + add_edge( + source_id, + _make_id(str(path_by_resolved.get(origin[0], origin[0]))), + "re_exports", + "export", + export_fact.line, + export_fact.file_path, + ) + + def resolve_exported_origin(target_path: Path, imported_name: str, seen: set[tuple[Path, str]] | None = None) -> tuple[Path, str]: + target_path = target_path.resolve() + key = (target_path, imported_name) + if seen is None: + seen = set() + if key in seen: + return key + seen.add(key) + origin = named_exports_by_file.get(target_path, {}).get(imported_name) + if origin is not None: + return resolve_exported_origin(origin[0], origin[1], seen) + for star_target in star_exports_by_file.get(target_path, []): + star_key = (star_target, imported_name) + if star_key in symbol_nodes: + return star_key + resolved = resolve_exported_origin(star_target, imported_name, seen) + if resolved in symbol_nodes: + return resolved + return key + + for import_fact in facts.imports: + source_id = source_file_id.get(import_fact.file_path.resolve()) + if source_id is None: + continue + origin_path, origin_symbol = resolve_exported_origin( + import_fact.target_path, + import_fact.imported_name, + ) + target_id = symbol_nodes.get((origin_path, origin_symbol)) + if target_id is None: + continue + add_edge( + source_id, + target_id, + "imports", + "import", + import_fact.line, + import_fact.file_path, + ) + + # #1146: emit file-to-file imports_from edges for package-form submodule imports. + for from_path, to_path, line in facts.module_imports: + try: + from_rel = from_path.relative_to(root) + to_rel = to_path.relative_to(root) + except ValueError: + continue + source_id = _make_id(_file_stem(from_rel)) + target_id = _make_id(_file_stem(to_rel)) + add_edge(source_id, target_id, "imports_from", "submodule_import", line, from_path) + + for use_fact in facts.uses: + file_path = use_fact.file_path.resolve() + target_id = None + unresolved_origin = local_aliases_by_file.get(file_path, {}).get(use_fact.local_name) + if unresolved_origin is not None: + origin_path, origin_symbol = resolve_exported_origin(*unresolved_origin) + target_id = symbol_nodes.get((origin_path, origin_symbol)) + if target_id is None and use_fact.relation in ("inherits", "implements"): + # Same-file fallback for HERITAGE only: a base declared in the same + # file (`class X extends Y`, `interface A extends B`) has no import + # alias, so resolve it directly against the file's own symbol nodes. + # Scoped to heritage because same-file calls/uses already resolve via + # the dedicated call-graph pass; widening this would duplicate those + # edges. Import resolution still takes precedence (#1095). + target_id = symbol_nodes.get((file_path, use_fact.local_name)) + if target_id is None: + continue + add_edge( + use_fact.source_id, + target_id, + use_fact.relation, + use_fact.context, + use_fact.line, + use_fact.file_path, + ) + +def _parse_js_tree(path: Path): + try: + from tree_sitter import Language, Parser + # .vue embeds the script in non-JS markup; mask it out and parse the + #