skills/graphify-labs__graphify
1
0
Fork 0
mirror of https://github.com/Graphify-Labs/graphify.git synced 2026-09-14 19:34:09 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
483a1d30a3836136cdac37c9e72b01914d95a6c2
graphify-labs__graphify/graphify/extractors/resolution.py
T
Vasu Bansal 483a1d30a3 fix(resolution): require owned sources for type-use edges 
(cherry picked from commit dcd9a7ee2d)
2026-09-05 21:13:54 +01:00

3227 lines
134 KiB
Python
Raw Blame History

"""resolution — moved verbatim from graphify/extract.py."""
from __future__ import annotations
from typing import Any, Callable
from pathlib import Path
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.base import ( # noqa: F401
_LANGUAGE_BUILTIN_GLOBALS,
_file_stem,
_make_id,
_read_text,
)
import hashlib
import json
import os
import re
import sys
_TSCONFIG_ALIAS_CACHE: dict[str, dict[str, list[str]]] = {}
# compilerOptions.baseUrl per config path, as an absolute dir (#2153).
_TSCONFIG_BASEURL_CACHE: "dict[str, Path | None]" = {}
_WORKSPACE_MANIFEST_NAMES = ("pnpm-workspace.yaml", "package.json")
_JS_RESOLVE_EXTS = (".ts", ".tsx", ".mts", ".cts", ".svelte", ".js", ".jsx", ".mjs", ".cjs")
_JS_INDEX_FILES = ("index.ts", "index.tsx", "index.svelte", "index.js", "index.jsx", "index.mjs")
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 <dir>/src/services rather
# than <dir>/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 _read_json_config(path: Path) -> "dict | None":
"""Parse a tsconfig/jsconfig as JSON, falling back to JSONC (#2153).
Mirrors the read/parse handling in `_read_tsconfig_aliases`; returns None on
any unreadable or unparseable file so a malformed config degrades to "no
baseUrl" instead of raising.
"""
try:
raw = path.read_text(encoding="utf-8", errors="replace")
except OSError:
return None
for candidate in (raw, _strip_jsonc(raw)):
try:
data = json.loads(candidate)
except Exception:
continue
return data if isinstance(data, dict) else None
return None
def _find_js_config(start_dir: Path) -> "tuple[Path, Path] | None":
"""Nearest tsconfig.json/jsconfig.json walking up from start_dir.
`jsconfig.json` is the plain-JS spelling of the same file (already indexed by
json_config.py) and was never probed here, so a Rails/webpacker project that
configures resolution in jsconfig.json got no aliases at all (#2153).
tsconfig.json wins when both sit in one directory, matching tsc and editors,
which consult jsconfig.json only when there is no tsconfig.json.
"""
current = start_dir.resolve()
for candidate in [current, *current.parents]:
for name in ("tsconfig.json", "jsconfig.json"):
config = candidate / name
if config.exists():
return config, candidate
return None
def _load_tsconfig_aliases(start_dir: Path) -> dict[str, list[str]]:
"""Walk up from start_dir to find tsconfig/jsconfig.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 config path string. The cache has no mtime/content
component, so extract() clears it per run (#2917); do not assume the entry
survives a config edit within a long-lived process.
"""
found = _find_js_config(start_dir)
if found is None:
return {}
config, candidate = found
key = str(config)
if key not in _TSCONFIG_ALIAS_CACHE:
_TSCONFIG_ALIAS_CACHE[key] = _read_tsconfig_aliases(config, candidate, seen=set())
return _TSCONFIG_ALIAS_CACHE[key]
def _load_tsconfig_base_url(start_dir: Path) -> "Path | None":
"""`compilerOptions.baseUrl` of the nearest config, as an absolute directory.
baseUrl was only ever used as the base that `paths` targets resolve against,
so a config declaring baseUrl and NO paths yielded an empty alias map and
every non-relative import went unresolved (#2153). Exposed separately so it
can act as a resolution root of last resort, after all declared aliases miss.
Returns None when no config declares baseUrl. Cached by config path with no
mtime component, so extract() clears it per run (#2917).
"""
found = _find_js_config(start_dir)
if found is None:
return None
config, candidate = found
key = str(config)
if key not in _TSCONFIG_BASEURL_CACHE:
base_url = None
data = _read_json_config(config)
if data is not None:
raw_base = data.get("compilerOptions", {}).get("baseUrl")
if isinstance(raw_base, str) and raw_base:
base_url = Path(os.path.normpath(candidate / raw_base))
_TSCONFIG_BASEURL_CACHE[key] = base_url
return _TSCONFIG_BASEURL_CACHE[key]
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]],
base_url: "Path | None" = None) -> "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.
`base_url` is a resolution root of last resort, tried only when NO declared
alias matches (#2153). It must not participate in the specificity contest:
a bare `*` alias would score (1, 0) and so beat a declared non-wildcard
directory-prefix alias at (2, -len), silently shadowing it and regressing
#1269. Unlike the alias path this returns a candidate only when it is a real
file on disk, so a genuine external package (`import React from 'react'`)
still resolves to nothing instead of a fabricated <baseUrl>/react edge.
"""
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:
if base_url is not None:
candidate = Path(os.path.normpath(base_url / raw))
resolved = _resolve_js_import_path(candidate)
if resolved.is_file():
return resolved
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
_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,
base_url=_load_tsconfig_base_url(start_dir))
if hit is not None:
return _resolve_js_import_path(hit)
return _resolve_workspace_import(raw, start_dir)
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 _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 _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)
_VUE_SCRIPT_RE = re.compile(
r"""(<script\b(?:"[^"]*"|'[^']*'|[^>"'])*>)([\s\S]*?)(</script\s*>)""",
re.IGNORECASE,
)
_VUE_SCRIPT_LANG_RE = re.compile(
r"""\blang\s*=\s*['"]?([A-Za-z]+)['"]?""", re.IGNORECASE
)
def _vue_mask_non_script(src: str) -> tuple[str, str | None]:
"""Blank everything outside ``<script>`` bodies, keeping ``\\r``/``\\n``.
Replaces template/style/tags with spaces so a JS/TS grammar sees only the
script, while preserved newlines keep line numbers accurate. Returns
``(masked_source, lang)``; ``lang`` is the first block's declared ``lang``.
"""
def _blank(s: str) -> str:
return re.sub(r"[^\r\n]", " ", s)
out: list[str] = []
pos = 0
lang: str | None = None
for m in _VUE_SCRIPT_RE.finditer(src):
out.append(_blank(src[pos:m.start()])) # markup/style before this block
out.append(_blank(m.group(1))) # <script …> open tag
out.append(m.group(2)) # script body, verbatim
out.append(_blank(m.group(3))) # </script> 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:
# No colliding ids to salt apart, but the transient `target_file` hint an
# importer stamps on every resolved import (#1814) still has to be dropped
# here — this early exit skips the edge loop below, so without it a
# non-colliding import would carry its absolute path into graph.json.
for edge in edges:
edge.pop("target_file", None)
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)
# An import/re-export edge's target is a FILE node that can collapse with a
# same-basename cross-extension sibling (foo.ts vs foo.mjs, #1814). Keying
# its target salt by the IMPORTER's own source_file mis-points it back at the
# importer's variant (a self-loop). When the emitter stamped the resolved
# target file, key the target salt by THAT file so the salt lands on the
# correct sibling. Generalizes the #1475 C/ObjC header carve-out (below) to
# every language and to re_exports. `pop` it as we consume it: this is the
# hint's only reader, and its absolute path must not persist into graph.json.
target_file = edge.pop("target_file", None)
if target_file and edge.get("relation") in ("imports", "imports_from", "re_exports"):
target_edge_key = _source_key(str(target_file), root)
else:
target_edge_key = edge_source_key
target_key = (edge.get("target", ""), target_edge_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, target_file: str | None = None, local_alias: str | None = None, type_only: bool = False) -> None:
key = (source, target, relation, context or "")
if key in existing_edges:
return
existing_edges.add(key)
edge = {
"source": source,
"target": target,
"relation": relation,
"context": context,
"confidence": "EXTRACTED",
"source_file": str(source_path),
"source_location": f"L{line}",
"weight": 1.0,
}
# A re-export edge's target is a FILE node that can collapse with a
# same-basename cross-extension sibling; stamp the resolved target file so
# the id-disambiguation salt is keyed by the TARGET, not the importer (#1814).
if target_file is not None:
edge["target_file"] = target_file
# The local name this import bound in the importing file, when it differs
# from the target's own name (`from pkg import mod as alias`) -- lets the
# cross-file member-call resolver match `alias.func()` (#2082).
if local_alias is not None:
edge["local_alias"] = local_alias
# Erased at compile time (#3123): Import Cycles skips these edges.
if type_only:
edge["type_only"] = True
edges.append(edge)
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,
target_file=str(path_by_resolved.get(target_path, target_path)),
type_only=star_fact.type_only,
)
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,
target_file=str(path_by_resolved.get(target_path, target_path)),
type_only=namespace_fact.type_only,
)
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,
target_file=str(path_by_resolved.get(origin[0], origin[0])),
type_only=export_fact.type_only,
)
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, local_name 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,
local_alias=local_name if local_name != to_path.stem else None,
)
# #2262 producer guard: never emit a `calls` use-edge from a source id
# that owns no node. All node appends (ensure_symbol_node, declarations,
# namespace exports) happened above, so the owned set is complete here.
# A node-less caller id can never be canonicalized by the extract()
# remaps (they learn only from nodes), so an absolute-derived one would
# leak the machine/scan-path slug into the edge source. Reattribute the
# edge to the caller's FILE node — the true file-level dependency
# survives, and the file id is exactly what the #2231 remap
# canonicalizes — or drop it when no file node id is available.
owned = {str(n.get("id")) for n in nodes}
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
source_id = use_fact.source_id
# Structural walking can omit named-function-local declarations while
# materializing callback-local ones. Only actual node ownership decides
# whether a type relationship has a represented source declaration.
if use_fact.relation in ("inherits", "implements", "references") and source_id not in owned:
continue
if use_fact.relation == "calls" and source_id not in owned:
source_id = source_file_id.get(file_path)
if source_id is None:
continue
add_edge(
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
# <script> with TS.
vue_lang: str | None = None
if path.suffix == ".vue":
masked, vue_lang = _vue_mask_non_script(
path.read_text(encoding="utf-8", errors="replace")
)
source = masked.encode("utf-8")
else:
source = path.read_bytes()
use_ts = path.suffix in (".ts", ".mts", ".cts") or (
path.suffix == ".vue" and vue_lang not in ("js", "jsx")
)
if path.suffix == ".tsx":
# .tsx must use the JSX-aware TSX grammar, mirroring the engine's
# _TSX_CONFIG (ts_language_fn="language_tsx"). Parsing .tsx with
# language_typescript misparses JSX, and tree-sitter's error
# recovery floats nested arrow components up to top level —
# _js_top_level_function_bodies then mints caller ids for callers
# that own no node, leaking absolute-path slugs into calls-edge
# sources (#2262).
import tree_sitter_typescript as tstypescript
language = Language(tstypescript.language_tsx())
elif use_ts:
import tree_sitter_typescript as tstypescript
language = Language(tstypescript.language_typescript())
else:
import tree_sitter_javascript as tsjavascript
language = Language(tsjavascript.language())
parser = Parser(language)
return source, parser.parse(source).root_node
except Exception:
return None
def _walk_js_tree(node):
# Iterative DFS avoids Python's O(depth) generator-chain overhead.
# Recursive yield-from creates one generator frame per level — at 26+
# levels deep each leaf's value had to propagate through 26 frames.
stack = [node]
while stack:
n = stack.pop()
yield n
stack.extend(reversed(n.children))
def _js_module_specifier(node, source: bytes) -> str | None:
source_node = node.child_by_field_name("source")
if source_node is None:
for child in node.children:
if child.type == "string":
source_node = child
break
if source_node is None:
return None
raw = _read_text(source_node, source).strip()
return raw.strip("'\"`") or None
def _js_named_specifiers(node, source: bytes, specifier_type: str) -> list[tuple[str, str]]:
pairs: list[tuple[str, str]] = []
for child in _walk_js_tree(node):
if child.type != specifier_type:
continue
name_node = child.child_by_field_name("name")
if name_node is None:
continue
alias_node = child.child_by_field_name("alias")
name = _read_text(name_node, source)
exposed = _read_text(alias_node, source) if alias_node is not None else name
if name and exposed:
pairs.append((name, exposed))
return pairs
def _js_export_clause(node):
for child in node.children:
if child.type == "export_clause":
return child
return None
def _js_export_statement_is_star(node) -> bool:
return any(child.type == "*" for child in node.children)
def _js_namespace_export_name(node, source: bytes) -> str | None:
for child in node.children:
if child.type != "namespace_export":
continue
for sub in child.children:
if sub.type == "identifier":
return _read_text(sub, source) or None
return None
def _js_lexical_aliases(node, source: bytes) -> list[tuple[str, str]]:
aliases: list[tuple[str, str]] = []
if node.type != "lexical_declaration":
return aliases
for child in node.children:
if child.type != "variable_declarator":
continue
name_node = child.child_by_field_name("name")
value_node = child.child_by_field_name("value")
if (
name_node is not None
and value_node is not None
and value_node.type in ("identifier", "type_identifier")
):
aliases.append((_read_text(name_node, source), _read_text(value_node, source)))
return aliases
def _js_exported_declaration_names(node, source: bytes) -> list[str]:
names: list[str] = []
declaration = node.child_by_field_name("declaration")
if declaration is None:
return names
if declaration.type == "lexical_declaration":
names.extend(alias for alias, _target in _js_lexical_aliases(declaration, source))
return names
if declaration.type in (
"class_declaration",
"abstract_class_declaration",
"interface_declaration",
"type_alias_declaration",
"function_declaration",
):
name_node = declaration.child_by_field_name("name")
if name_node is not None:
names.append(_read_text(name_node, source))
return names
def _js_default_import_name(node, source: bytes) -> str | None:
"""Local binding of a default import: the `Foo` in `import Foo from './x'`.
The default binding is a bare identifier child of the import_clause (named
imports live in a `named_imports` node, namespace imports in a
`namespace_import` node), so it is also picked up from the mixed form
`import Foo, { Bar } from './x'`.
"""
for child in node.children:
if child.type == "import_clause":
for sub in child.children:
if sub.type == "identifier":
return _read_text(sub, source)
return None
def _js_default_export_name(node, source: bytes) -> str | None:
"""Local name of a default export, or None for anonymous defaults.
Handles `export default class Foo {}`, `export default function foo() {}`,
`export default abstract class Foo {}` (name on the `declaration` field) and
`export default Foo` (an identifier on the `value` field). Anonymous defaults
(`export default class {}`, `export default {...}`) have no resolvable symbol
and return None.
"""
if not any(child.type == "default" for child in node.children):
return None
declaration = node.child_by_field_name("declaration")
if declaration is not None:
name_node = declaration.child_by_field_name("name")
return _read_text(name_node, source) if name_node is not None else None
value = node.child_by_field_name("value")
if value is not None and value.type == "identifier":
return _read_text(value, source)
return None
def _js_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]:
bodies: list[tuple[str, object]] = []
stem = _file_stem(path)
for node in root_node.children:
if node.type == "function_declaration":
name_node = node.child_by_field_name("name")
body = node.child_by_field_name("body")
if name_node is not None and body is not None:
bodies.append((_make_id(stem, _read_text(name_node, source)), body))
continue
if node.type != "lexical_declaration":
continue
for child in node.children:
if child.type != "variable_declarator":
continue
name_node = child.child_by_field_name("name")
value_node = child.child_by_field_name("value")
if (
name_node is not None
and value_node is not None
and value_node.type == "arrow_function"
):
bodies.append((_make_id(stem, _read_text(name_node, source)), value_node))
return bodies
def _js_call_identifier(node, source: bytes) -> str | None:
if node.type != "call_expression":
return None
function_node = node.child_by_field_name("function")
if function_node is None:
for child in node.children:
if child.is_named:
function_node = child
break
if function_node is not None and function_node.type in ("identifier", "type_identifier"):
return _read_text(function_node, source)
return None
_JS_PRIMITIVE_TYPES = frozenset({
"string", "number", "boolean", "any", "unknown", "void", "never",
"object", "null", "undefined", "bigint", "symbol", "this",
})
def _ts_heritage_clause_entries(clause_node, source: bytes) -> list[str]:
"""Return base/interface type names from an extends_clause or implements_clause."""
out: list[str] = []
for child in clause_node.children:
if not child.is_named:
continue
if child.type in ("identifier", "type_identifier"):
name = _read_text(child, source)
if name:
out.append(name)
elif child.type == "generic_type":
name_node = child.child_by_field_name("name")
if name_node is None:
for sub in child.children:
if sub.type in ("type_identifier", "nested_type_identifier", "identifier"):
name_node = sub
break
if name_node is not None:
text = _read_text(name_node, source).rsplit(".", 1)[-1]
if text:
out.append(text)
elif child.type == "nested_type_identifier":
text = _read_text(child, source).rsplit(".", 1)[-1]
if text:
out.append(text)
return out
def _ts_collect_type_refs(node, source: bytes, generic: bool, out: list[tuple[str, str]]) -> None:
"""Walk a TS type annotation tree; append (name, role) tuples.
role is 'type' for the outermost type position and 'generic_arg' for entries
that appear inside `type_arguments`.
"""
if node is None:
return
t = node.type
if t == "type_annotation":
for c in node.children:
if c.is_named:
_ts_collect_type_refs(c, source, generic, out)
return
if t in ("type_identifier", "identifier"):
name = _read_text(node, source)
if name and name not in _JS_PRIMITIVE_TYPES:
out.append((name, "generic_arg" if generic else "type"))
return
if t == "nested_type_identifier":
tail = _read_text(node, source).rsplit(".", 1)[-1]
if tail and tail not in _JS_PRIMITIVE_TYPES:
out.append((tail, "generic_arg" if generic else "type"))
return
if t == "generic_type":
name_node = node.child_by_field_name("name")
if name_node is not None:
text = _read_text(name_node, source).rsplit(".", 1)[-1]
if text and text not in _JS_PRIMITIVE_TYPES:
out.append((text, "generic_arg" if generic else "type"))
else:
for c in node.children:
if c.type in ("type_identifier", "nested_type_identifier"):
text = _read_text(c, source).rsplit(".", 1)[-1]
if text and text not in _JS_PRIMITIVE_TYPES:
out.append((text, "generic_arg" if generic else "type"))
break
for c in node.children:
if c.type == "type_arguments":
for sub in c.children:
if sub.is_named:
_ts_collect_type_refs(sub, source, True, out)
return
if node.is_named:
for c in node.children:
if c.is_named:
_ts_collect_type_refs(c, source, generic, out)
def _ts_walk_class_members(class_node, source: bytes, path: Path, class_nid: str,
facts: _SymbolResolutionFacts) -> None:
"""Emit type-relation and type-reference use facts for a class declaration node."""
line = class_node.start_point[0] + 1
for child in class_node.children:
if child.type == "class_heritage":
saw_clause = False
for clause in child.children:
if clause.type == "extends_clause":
saw_clause = True
for name in _ts_heritage_clause_entries(clause, source):
facts.uses.append(
_SymbolUseFact(path, class_nid, name, "inherits", "type",
clause.start_point[0] + 1)
)
elif clause.type == "implements_clause":
saw_clause = True
for name in _ts_heritage_clause_entries(clause, source):
facts.uses.append(
_SymbolUseFact(path, class_nid, name, "implements", "type",
clause.start_point[0] + 1)
)
if not saw_clause:
# The JavaScript grammar carries the base directly under
# class_heritage (`extends Base` -> [extends, identifier]) with no
# extends_clause/implements_clause wrapper like the TypeScript
# grammar. Treat the heritage node itself as the extends clause so
# `class Derived extends Base {}` in a .js file still emits an
# inherits edge. Mirrors the extends_type_clause branch below.
for name in _ts_heritage_clause_entries(child, source):
facts.uses.append(
_SymbolUseFact(path, class_nid, name, "inherits", "type",
child.start_point[0] + 1)
)
elif child.type == "extends_type_clause":
# Interface heritage (`interface A extends B, C`) is an
# extends_type_clause node, NOT a class_heritage. Its base entries
# are the same node types extends_clause holds, so the helper is
# reusable. Without this branch interface inheritance is dropped (#1095).
for name in _ts_heritage_clause_entries(child, source):
facts.uses.append(
_SymbolUseFact(path, class_nid, name, "inherits", "type",
child.start_point[0] + 1)
)
body = class_node.child_by_field_name("body")
if body is None:
return
for member in body.children:
m_line = member.start_point[0] + 1
if member.type in ("method_definition", "method_signature", "abstract_method_signature"):
name_node = member.child_by_field_name("name")
if name_node is None:
continue
method_name = _read_text(name_node, source)
method_nid = _make_id(class_nid, method_name)
params = member.child_by_field_name("parameters")
if params is not None:
for p in params.children:
if p.type not in ("required_parameter", "optional_parameter"):
continue
type_anno = p.child_by_field_name("type")
if type_anno is None:
continue
refs: list[tuple[str, str]] = []
_ts_collect_type_refs(type_anno, source, False, refs)
for name, role in refs:
ctx = "generic_arg" if role == "generic_arg" else "parameter_type"
facts.uses.append(
_SymbolUseFact(path, method_nid, name, "references", ctx, m_line)
)
return_type = member.child_by_field_name("return_type")
if return_type is not None:
refs = []
_ts_collect_type_refs(return_type, source, False, refs)
for name, role in refs:
ctx = "generic_arg" if role == "generic_arg" else "return_type"
facts.uses.append(
_SymbolUseFact(path, method_nid, name, "references", ctx, m_line)
)
elif member.type in ("public_field_definition", "property_signature"):
type_anno = None
for c in member.children:
if c.type == "type_annotation":
type_anno = c
break
if type_anno is None:
continue
refs = []
_ts_collect_type_refs(type_anno, source, False, refs)
for name, role in refs:
ctx = "generic_arg" if role == "generic_arg" else "field"
facts.uses.append(
_SymbolUseFact(path, class_nid, name, "references", ctx, m_line)
)
def _collect_js_symbol_resolution_facts(paths: list[Path], facts: _SymbolResolutionFacts) -> None:
js_paths = [
path for path in paths
if path.suffix in _JS_CACHE_BYPASS_SUFFIXES
]
if not js_paths:
return
trees: dict[Path, tuple[bytes, object]] = {}
for path in js_paths:
resolved_path = path.resolve()
parsed = _parse_js_tree(path)
if parsed is None:
continue
source, root_node = parsed
trees[resolved_path] = parsed
for node in _walk_js_tree(root_node):
if node.type == "export_statement":
for name in _js_exported_declaration_names(node, source):
facts.declarations.append(
_SymbolDeclarationFact(path, name, node.start_point[0] + 1)
)
if node.type != "import_statement":
continue
raw_module = _js_module_specifier(node, source)
if raw_module is None:
continue
target_path = _resolve_js_module_path(raw_module, path.parent)
if target_path is None:
continue
target_path = target_path.resolve()
for imported_name, local_name in _js_named_specifiers(node, source, "import_specifier"):
facts.imports.append(
_SymbolImportFact(
path,
local_name,
target_path,
imported_name,
node.start_point[0] + 1,
)
)
default_local = _js_default_import_name(node, source)
if default_local is not None:
facts.imports.append(
_SymbolImportFact(
path,
default_local,
target_path,
"default",
node.start_point[0] + 1,
)
)
for node in _walk_js_tree(root_node):
for alias, target in _js_lexical_aliases(node, source):
facts.aliases.append(
_SymbolAliasFact(path, alias, target, node.start_point[0] + 1)
)
for path in js_paths:
resolved_path = path.resolve()
parsed = trees.get(resolved_path)
if parsed is None:
continue
source, root_node = parsed
for node in _walk_js_tree(root_node):
if node.type != "export_statement":
continue
raw_module = _js_module_specifier(node, source)
export_clause = _js_export_clause(node)
# `export type { X } from ...` / `export type * from ...`: the
# statement-level `type` keyword is a bare anonymous child; the
# default binding NAMED type sits inside the clause instead (#3123).
stmt_type_only = any(
child.type == "type" and not child.is_named
for child in node.children
)
if raw_module is not None:
target_path = _resolve_js_module_path(raw_module, path.parent)
if target_path is None:
continue
target_path = target_path.resolve()
namespace_name = _js_namespace_export_name(node, source)
if namespace_name is not None:
facts.namespace_exports.append(
_NamespaceExportFact(
path,
namespace_name,
target_path,
node.start_point[0] + 1,
type_only=stmt_type_only,
)
)
elif _js_export_statement_is_star(node):
facts.star_exports.append(
_StarExportFact(path, target_path, node.start_point[0] + 1,
type_only=stmt_type_only)
)
if export_clause is not None:
for original_name, exported_name in _js_named_specifiers(
export_clause, source, "export_specifier"
):
facts.exports.append(
_SymbolExportFact(
path,
exported_name,
node.start_point[0] + 1,
target_path=target_path,
target_name=original_name,
type_only=stmt_type_only,
)
)
continue
if export_clause is not None:
for local_name, exported_name in _js_named_specifiers(
export_clause, source, "export_specifier"
):
facts.exports.append(
_SymbolExportFact(
path,
exported_name,
node.start_point[0] + 1,
local_name=local_name,
)
)
continue
for exported_name in _js_exported_declaration_names(node, source):
facts.exports.append(
_SymbolExportFact(
path,
exported_name,
node.start_point[0] + 1,
local_name=exported_name,
)
)
# `export default class Foo {}` / `export default foo` exposes the
# symbol under the name "default"; record that so a default import
# (imported_name="default") resolves to it. `export { X as default }`
# is already handled via the export_clause path above.
default_name = _js_default_export_name(node, source)
if default_name is not None:
facts.exports.append(
_SymbolExportFact(
path,
"default",
node.start_point[0] + 1,
local_name=default_name,
)
)
for path in js_paths:
resolved_path = path.resolve()
parsed = trees.get(resolved_path)
if parsed is None:
continue
source, root_node = parsed
for source_id, body in _js_top_level_function_bodies(path, root_node, source):
for node in _walk_js_tree(body):
imported_name = _js_call_identifier(node, source)
if imported_name is None:
continue
facts.uses.append(
_SymbolUseFact(
path,
source_id,
imported_name,
"calls",
"call",
node.start_point[0] + 1,
)
)
for path in js_paths:
resolved_path = path.resolve()
parsed = trees.get(resolved_path)
if parsed is None:
continue
source, root_node = parsed
stem = _file_stem(path)
for node in _walk_js_tree(root_node):
if node.type not in (
"class_declaration",
"abstract_class_declaration",
"interface_declaration",
):
continue
name_node = node.child_by_field_name("name")
if name_node is None:
continue
class_name = _read_text(name_node, source)
if not class_name:
continue
class_nid = _make_id(stem, class_name)
_ts_walk_class_members(node, source, path, class_nid, facts)
def _parse_python_tree(path: Path):
try:
import tree_sitter_python as tspython
from tree_sitter import Language, Parser
source = path.read_bytes()
parser = Parser(Language(tspython.language()))
return source, parser.parse(source).root_node
except Exception:
return None
def _walk_python_tree(node):
yield node
for child in node.children:
yield from _walk_python_tree(child)
def _python_import_from_module(node, source: bytes) -> tuple[int, str] | None:
level = 0
module_name = ""
for child in node.children:
if child.type == "import":
break
if child.type == "relative_import":
raw = _read_text(child, source)
level = len(raw) - len(raw.lstrip("."))
remainder = raw.lstrip(".")
if remainder:
module_name = remainder
for sub in child.children:
if sub.type == "dotted_name":
module_name = _read_text(sub, source)
elif child.type == "dotted_name":
module_name = _read_text(child, source)
if level == 0 and not module_name:
return None
return level, module_name
def _python_imported_names(node, source: bytes) -> list[tuple[str, str]]:
names: list[tuple[str, str]] = []
past_import = False
for child in node.children:
if child.type == "import":
past_import = True
continue
if not past_import:
continue
if child.type == "dotted_name":
name = _read_text(child, source)
names.append((name, name.split(".")[-1]))
elif child.type == "aliased_import":
name_node = child.child_by_field_name("name")
alias_node = child.child_by_field_name("alias")
if name_node is None:
continue
name = _read_text(name_node, source)
local = _read_text(alias_node, source) if alias_node is not None else name.split(".")[-1]
names.append((name, local))
return names
def _probe_python_module_candidate(candidate: Path) -> Path | None:
"""Resolve one module-path candidate to a .py file (dir+__init__, exact, or
with a .py suffix), or None."""
if candidate.is_dir():
init_path = candidate / "__init__.py"
if init_path.is_file():
return init_path
if candidate.is_file():
return candidate
if not candidate.name:
return None
py_candidate = candidate.with_suffix(".py")
if py_candidate.is_file():
return py_candidate
return None
def _resolve_python_module_path(module_name: str, current_path: Path, root: Path, level: int) -> Path | None:
if level > 0:
base = current_path.parent
for _ in range(level - 1):
base = base.parent
candidate = base / module_name.replace(".", "/") if module_name else base
return _probe_python_module_candidate(candidate)
# Absolute import. Probe the scan root first (unchanged for the common
# root-is-package-root layout), then walk up from the importing file toward
# the root so a `src/` (or otherwise nested) package root resolves regardless
# of where the scan started — `import pkg.mod` from src/pkg/app.py must find
# src/pkg/mod.py whether the scan root is the repo or src/ (#2072). Mirrors
# the upward walk already used for Lua (_resolve_lua_import_target, #1075).
rel = module_name.replace(".", "/")
hit = _probe_python_module_candidate(root / rel)
if hit is not None:
return hit
for anc in current_path.parents:
try:
anc.relative_to(root)
except ValueError:
break # left the scan root; stop walking up
if anc == root:
continue # already probed root/rel above
# Only probe sys.path-root candidates — dirs that are NOT themselves part
# of a package. Probing a package dir would resolve an absolute
# `from helpers import x` to a sibling in the current package (Python-2
# implicit-relative semantics), fabricating edges to what may be an
# external dependency (#2072 review). A src-layout root (src/, no
# __init__.py) is still probed.
if (anc / "__init__.py").is_file():
continue
cand = _probe_python_module_candidate(anc / rel)
if cand is not None:
return cand
return None
def _python_top_level_function_bodies(path: Path, root_node, source: bytes) -> list[tuple[str, object]]:
bodies: list[tuple[str, object]] = []
stem = _file_stem(path)
for node in root_node.children:
if node.type != "function_definition":
continue
name_node = node.child_by_field_name("name")
body = node.child_by_field_name("body")
if name_node is not None and body is not None:
bodies.append((_make_id(stem, _read_text(name_node, source)), body))
return bodies
def _python_call_identifier(node, source: bytes) -> str | None:
if node.type != "call":
return None
function_node = node.child_by_field_name("function")
if function_node is not None and function_node.type == "identifier":
return _read_text(function_node, source)
return None
def _collect_python_symbol_resolution_facts(
paths: list[Path],
root: Path,
facts: _SymbolResolutionFacts,
) -> None:
py_paths = [path for path in paths if path.suffix == ".py"]
if not py_paths:
return
trees: dict[Path, tuple[bytes, object]] = {}
for path in py_paths:
parsed = _parse_python_tree(path)
if parsed is None:
continue
source, root_node = parsed
trees[path.resolve()] = parsed
for node in _walk_python_tree(root_node):
if node.type != "import_from_statement":
continue
module = _python_import_from_module(node, source)
if module is None:
continue
level, module_name = module
target_path = _resolve_python_module_path(module_name, path, root, level)
if target_path is None:
continue
# #1146: `from pkg import submod` — if the target is a package
# (__init__.py) and an imported name matches a submodule file on
# disk, emit a file-level import edge to that submodule rather
# than only to the package.
pkg_dir = target_path.parent if target_path.name == "__init__.py" else None
for imported_name, local_name in _python_imported_names(node, source):
line = node.start_point[0] + 1
if pkg_dir is not None:
sub_py = pkg_dir / f"{imported_name}.py"
sub_pkg = pkg_dir / imported_name / "__init__.py"
submodule = sub_py if sub_py.is_file() else (sub_pkg if sub_pkg.is_file() else None)
if submodule is not None:
facts.module_imports.append((path, submodule, line, local_name))
continue
facts.imports.append(
_SymbolImportFact(path, local_name, target_path, imported_name, line)
)
if path.name == "__init__.py":
facts.exports.append(
_SymbolExportFact(
path,
local_name,
line,
target_path=target_path,
target_name=imported_name,
)
)
for path in py_paths:
parsed = trees.get(path.resolve())
if parsed is None:
continue
source, root_node = parsed
for source_id, body in _python_top_level_function_bodies(path, root_node, source):
for node in _walk_python_tree(body):
imported_name = _python_call_identifier(node, source)
if imported_name is None:
continue
facts.uses.append(
_SymbolUseFact(
path,
source_id,
imported_name,
"calls",
"call",
node.start_point[0] + 1,
)
)
def _augment_symbol_resolution_edges(
paths: list[Path],
nodes: list[dict],
edges: list[dict],
root: Path,
) -> None:
facts = _SymbolResolutionFacts()
_collect_js_symbol_resolution_facts(paths, facts)
_collect_python_symbol_resolution_facts(paths, root, facts)
_apply_symbol_resolution_facts(paths, nodes, edges, root, facts)
def _resolve_cross_file_imports(
per_file: list[dict],
paths: list[Path],
all_nodes: list[dict] | None = None,
all_edges: list[dict] | None = None,
) -> list[dict]:
"""
Two-pass import resolution: turn file-level imports into class-level edges.
Pass 1 - build a global map: class/function name → node_id, per stem.
Pass 2 - for each `from .module import Name`, look up Name in the global
map and add a direct INFERRED edge from each class in the
importing file to the imported entity.
This turns:
auth.py --imports_from--> models.py (obvious, filtered out)
Into:
DigestAuth --uses--> Response [INFERRED] (cross-file, interesting!)
BasicAuth --uses--> Request [INFERRED]
"""
try:
import tree_sitter_python as tspython
from tree_sitter import Language, Parser
except ImportError:
return []
language = Language(tspython.language())
parser = Parser(language)
# Pass 1: _file_stem(path) → {ClassName: node_id}
# Keyed by directory-qualified stem (e.g. "auth_models") to avoid collisions
# when multiple files share the same filename in different directories.
# A secondary bare-stem index handles absolute imports where only the module
# name is known — first writer wins when names collide (inherently ambiguous).
stem_to_entities: dict[str, dict[str, str]] = {}
bare_to_qualified: dict[str, str] = {}
for file_result in per_file:
for node in file_result.get("nodes", []):
src = node.get("source_file", "")
if not src:
continue
src_path = Path(src)
fq_stem = _file_stem(src_path)
label = node.get("label", "")
nid = node.get("id", "")
# Index class-level entities only. Function/method labels end in "()"
# so are excluded by the `endswith(")")` filter; file nodes end in ".py";
# private/internal labels start with "_"; rationale nodes carry
# file_type=="rationale" and must never participate in cross-file
# import resolution (#563).
if (
label
and not label.endswith((")", ".py"))
and "_" not in label[:1]
and node.get("file_type") != "rationale"
):
stem_to_entities.setdefault(fq_stem, {})[label] = nid
if src_path.stem not in bare_to_qualified:
bare_to_qualified[src_path.stem] = fq_stem
# Pass 2: for each file, find `from .X import A, B, C`, then attribute the
# `uses` edge to the specific local symbol (class OR function) whose body
# actually references the imported name — not to every class that merely
# shares the file (#2652). The edge is anchored at the real reference, not
# the import line, so `source_location` points at genuine corroboration.
new_edges: list[dict] = []
node_by_id = {node["id"]: node for node in all_nodes if node.get("id")} if all_nodes else {}
repointed_from: set[str] = set()
TYPE_REPOINT_RELATIONS = frozenset({"references", "inherits", "implements", "extends"})
for file_result, path in zip(per_file, paths):
str_path = str(path)
file_srcs = {n.get("source_file") for n in file_result.get("nodes", []) if n.get("source_file")}
file_srcs.add(str_path)
file_srcs.add(path.as_posix())
# Map each local symbol (class or function) to its node id, keyed by the
# bare symbol name. Function labels end in "()"; the file node ends in
# ".py"; rationale nodes never import (#563). First writer wins on a
# name collision (inherently ambiguous within a file).
name_to_nid: dict[str, str] = {}
for n in file_result.get("nodes", []):
if n.get("source_file") != str_path or n.get("file_type") == "rationale":
continue
label = n.get("label", "")
if not label or label.endswith(".py"):
continue
sym_name = label[:-2] if label.endswith("()") else label
if sym_name and sym_name not in name_to_nid:
name_to_nid[sym_name] = n["id"]
if not name_to_nid:
continue
# Parse imports from this file
try:
source = path.read_bytes()
tree = parser.parse(source)
except Exception:
continue
# local_name -> target node id (local_name honours `import X as Y`, so a
# reference to the alias in the body still attributes correctly).
import_targets: dict[str, str] = {}
# referenced name -> {source symbol nid: first reference line}
ref_sources: dict[str, dict[str, int]] = {}
def _text(n) -> str:
return source[n.start_byte:n.end_byte].decode("utf-8", errors="replace")
def resolve_import(node) -> None:
# Find the module name - handles both absolute and relative imports.
# Relative: `from .models import X` → relative_import → dotted_name
# Absolute: `from models import X` → module_name field
# target_fq is the directory-qualified stem used as the key in
# stem_to_entities. Relative imports are resolved exactly via the
# importing file's directory; absolute imports fall back to the
# bare-stem secondary index (first-writer-wins when names collide).
target_fq: str | None = None
for child in node.children:
if child.type == "relative_import":
prefix_text = ""
dotted_text = ""
for sub in child.children:
if sub.type == "import_prefix":
prefix_text = _text(sub)
elif sub.type == "dotted_name":
dotted_text = _text(sub)
dots = prefix_text.count(".") if prefix_text else 1
cur_dir = path.parent
for _ in range(dots - 1):
cur_dir = cur_dir.parent
if dotted_text:
candidate = cur_dir.joinpath(*dotted_text.split(".")).with_suffix(".py")
else:
candidate = cur_dir / "__init__.py"
target_fq = _file_stem(candidate)
break
if child.type == "dotted_name" and target_fq is None:
dotted_name = _text(child)
dotted_as_path = "/".join(dotted_name.split("."))
if dotted_as_path in stem_to_entities:
target_fq = dotted_as_path
else:
suffix_matches = [
fq for fq in stem_to_entities
if fq.endswith(f"/{dotted_as_path}") or fq.endswith(f"\\{dotted_as_path}")
]
if len(suffix_matches) == 1:
target_fq = suffix_matches[0]
else:
bare = dotted_name.split(".")[-1]
target_fq = bare_to_qualified.get(bare)
if not target_fq or target_fq not in stem_to_entities:
return
# Imported names come AFTER the 'import' keyword token. For
# `import X as Y` the target is found via X but the body uses Y.
past_import_kw = False
for child in node.children:
if child.type == "import":
past_import_kw = True
continue
if not past_import_kw:
continue
imported_name: str | None = None
local_name: str | None = None
if child.type == "dotted_name":
imported_name = local_name = _text(child)
elif child.type == "aliased_import":
name_node = child.child_by_field_name("name")
alias_node = child.child_by_field_name("alias")
if name_node is not None:
imported_name = _text(name_node)
local_name = _text(alias_node) if alias_node is not None else imported_name
if not imported_name or not local_name:
continue
tgt_nid = stem_to_entities[target_fq].get(imported_name)
if tgt_nid:
import_targets[local_name] = tgt_nid
def visit(node, current_nid: str | None) -> None:
# Identifiers inside an import statement are the import itself, not a
# real use — resolve the import here and don't descend into it.
if node.type == "import_from_statement":
resolve_import(node)
return
# Attribute references to the top-level symbol that contains them: a
# class is a unit (a reference inside one of its methods counts for
# the class, matching the documented DigestAuth->Response edge), and
# a module-level function is its own source. Only set at module scope
# (current_nid is None) so nested defs never override the container.
if current_nid is None and node.type in ("class_definition", "function_definition"):
name_node = node.child_by_field_name("name")
if name_node is not None:
mapped = name_to_nid.get(_text(name_node))
if mapped is not None:
current_nid = mapped
if node.type == "identifier" and current_nid is not None:
slot = ref_sources.setdefault(_text(node), {})
slot.setdefault(current_nid, node.start_point[0] + 1)
for child in node.children:
visit(child, current_nid)
visit(tree.root_node, None)
for name, tgt_nid in import_targets.items():
for src_nid, line in ref_sources.get(name, {}).items():
if src_nid == tgt_nid:
continue
new_edges.append({
"source": src_nid,
"target": tgt_nid,
"relation": "uses",
"confidence": "INFERRED",
# 0.95 = "direct structural evidence (named cross-file
# reference)" from the extraction-spec rubric, which is
# exactly what this edge is: a name this file imports, then
# references. Omitting the score entirely fell through to the
# 0.5 default the same rubric forbids outright (#2813).
"confidence_score": 0.95,
"source_file": str_path,
"source_location": f"L{line}",
"weight": 0.8,
})
# Repoint AST type-reference and inheritance edges from sourceless stubs
# to the exact imported target definition (#3252).
if all_edges and import_targets:
for edge in all_edges:
if edge.get("source_file") not in file_srcs:
continue
if edge.get("relation") not in TYPE_REPOINT_RELATIONS:
continue
tgt = edge.get("target")
tgt_node = node_by_id.get(tgt)
if not tgt_node or tgt_node.get("source_file"):
continue
stub_label = tgt_node.get("label", "")
resolved_id = import_targets.get(stub_label)
if resolved_id and resolved_id != tgt:
edge["target"] = resolved_id
repointed_from.add(tgt)
if all_nodes is not None and all_edges is not None and repointed_from:
still_referenced = {e.get("source") for e in all_edges} | {e.get("target") for e in all_edges}
all_nodes[:] = [
node for node in all_nodes
if node.get("id") not in repointed_from or node.get("id") in still_referenced
]
return new_edges
_DECLDEF_HEADER_SUFFIXES = frozenset({".h", ".hpp", ".hh", ".hxx"})
_DECLDEF_IMPL_SUFFIXES = frozenset({".m", ".mm", ".cpp", ".cc", ".cxx", ".c"})
def _decldef_class_stem(source_file: str) -> tuple[str, str] | None:
"""Return ``(dir, base_stem)`` for a header/impl source file, else None.
The base stem strips an ObjC category suffix (``Foo+Cat.m`` -> ``Foo``) so a
category implementation pairs with its ``Foo.h`` declaration. Files with an
extension that is neither a header nor an impl extension return None and are
never considered for the merge.
"""
if not source_file:
return None
p = Path(source_file)
suffix = p.suffix.lower()
if suffix not in _DECLDEF_HEADER_SUFFIXES and suffix not in _DECLDEF_IMPL_SUFFIXES:
return None
stem = p.stem.split("+", 1)[0] # ObjC category: Foo+Cat -> Foo
if not stem:
return None
return (str(p.parent), stem)
def _source_stem(node: dict) -> str:
"""Filename stem of a node's ``source_file`` (``""`` when it has none)."""
return Path(str(node.get("source_file", ""))).stem
def _merge_decl_def_classes(
all_nodes: list[dict],
all_edges: list[dict],
) -> None:
"""Merge a class (and its methods) declared in a header with its definition in
a sibling impl file into ONE node, for C/C++/ObjC (#1547, #1556).
A class declared in ``Foo.h`` (``class Foo`` / ``@interface Foo``) and defined
in the sibling ``Foo.cpp`` / ``Foo.m`` (``@implementation Foo``, plus — after
the C++ qualified-name fix — out-of-class method definitions ``Foo::bar``)
produces TWO nodes per symbol. Both are keyed off the file *stem*, and
``_file_stem`` drops the extension, so the header symbol and its impl
counterpart get the IDENTICAL id and differ only in ``source_file`` and label
(the C++ def label is ``Foo::bar()`` vs the decl's ``bar``; the ObjC impl class
label equals the interface's). Left alone, ``_disambiguate_colliding_node_ids``
SPLITS those id-collisions apart by path, fragmenting one class into two def
nodes — which then trips every resolver's single-definition god-node guard
(``len(defs) != 1`` -> bail), cascading into lost .h<->.m/.cpp linkage and dead
cross-file calls.
This pass runs BEFORE disambiguation and collapses each such id-collision to
ONE node — the header (declaration) variant, consistent with the #1475
header_remaps direction — so disambiguation sees a single source_file per id
and leaves it alone, and the downstream resolvers see ONE definition. Because
the colliding nodes already share an id, no edge re-pointing is needed: every
edge that referenced the impl symbol already points at the surviving id. We
only drop the redundant duplicate node and prefer the header's label — but
the dropped impl node's provenance is preserved on the survivor as
``definition_file`` / ``definition_location``, so the definition site is
still reachable from the merged node.
GOD-NODE GUARDS (false merges are the main risk):
* Collapse fires ONLY when every node in an id-collision group comes from a
SIBLING header/impl set — same directory, same base stem (ObjC categories
``Foo+Cat.m`` compare by the stem before ``+``), header extension paired
with impl extension — AND the group contains exactly ONE header file.
* Two unrelated ``class Logger`` in DIFFERENT directories never collide on id
(the id embeds the full file stem / directory path), so they are never
grouped and never merge. Two same-named classes in the SAME directory but
different base stems likewise key to different ids. Any id-collision that
is NOT a clean single-header sibling set is left untouched for
disambiguation to split (the conservative default).
The class and its method/field members fold in together: members are keyed
``_make_id(class_id, name)`` (ObjC) or, for an out-of-class C++ definition,
``_make_id(stem, "Foo::bar")`` which normalizes to the same id as the in-class
member ``_make_id(class_id, "bar")``. So every decl/def member pair is itself an
id-collision across the same sibling file set and collapses by the same rule.
"""
# Group every code node by id, recording the distinct source files involved.
by_id: dict[str, list[dict]] = {}
for n in all_nodes:
if n.get("file_type") != "code":
continue
nid = n.get("id")
sf = str(n.get("source_file", ""))
if not isinstance(nid, str) or not nid or not sf:
continue
by_id.setdefault(nid, []).append(n)
# Identify, per surviving id, which node to keep (header preferred). We can't
# mutate all_nodes mid-scan, so collect a set of node object ids to drop.
drop_objs: set[int] = set()
for nid, group in by_id.items():
if len(group) < 2:
continue
# The distinct source files of this collision must form a clean sibling
# header/impl set with exactly one header. Each file must parse as a
# header/impl file (others -> bail), share one directory + base stem.
sibling_keys: set[tuple[str, str]] = set()
headers: list[dict] = []
ok = True
for node in group:
sf = str(node.get("source_file", ""))
ds = _decldef_class_stem(sf)
if ds is None:
ok = False
break
sibling_keys.add(ds)
if Path(sf).suffix.lower() in _DECLDEF_HEADER_SUFFIXES:
headers.append(node)
if not ok:
continue
# All from one (dir, base_stem) sibling family. Pick the declaring header.
# Usually there is exactly one. An ObjC class whose members are split across
# categories has several (`Foo.h`, `Foo+Cat.h`) — fold those too, keeping the
# BASE header (the stem with no `+`), or the lowest-sorting category header
# when the base class lives outside the corpus (`NSString+Trim.h`). Two
# NON-category headers still bail to disambiguation, as before, so an
# unrelated `Foo.h` / `Foo.hpp` pair is untouched.
if len(sibling_keys) != 1 or not headers:
continue
if len(headers) == 1:
keeper = headers[0]
else:
base_headers = [h for h in headers if "+" not in _source_stem(h)]
if len(base_headers) > 1:
continue
keeper = base_headers[0] if base_headers else min(headers, key=_source_stem)
# The keeper is the DECLARATION, so without this the graph reports the
# header as the symbol's only location and the definition site — the file
# and line a reader actually wants — is discarded with the dropped node.
# Recorded as separate attributes: the survivor's id, label and
# source_file are untouched, so no existing graph is re-keyed, no edge
# moves, and the single-definition guarantee downstream resolvers rely on
# is unchanged. Chosen deterministically (lowest source_file, then
# location) so an ObjC class whose members are split across several
# category impls does not depend on node arrival order.
impls = sorted(
(n for n in group
if n is not keeper
and Path(str(n.get("source_file", ""))).suffix.lower()
in _DECLDEF_IMPL_SUFFIXES),
key=lambda n: (str(n.get("source_file", "")),
str(n.get("source_location", ""))),
)
if impls:
definition = impls[0]
if definition.get("source_file"):
keeper["definition_file"] = definition["source_file"]
if definition.get("source_location"):
keeper["definition_location"] = definition["source_location"]
for node in group:
if node is not keeper:
drop_objs.add(id(node))
if not drop_objs:
return
# Drop the redundant duplicate nodes. The surviving (header) node keeps its
# own label/source_file (and now carries the impl's definition_file/
# definition_location); edges are unchanged because the id is identical. Then
# de-dup any now-identical edges (e.g. the impl file's `contains`/`method`
# edge that duplicates the header's after the collapse).
all_nodes[:] = [n for n in all_nodes if id(n) not in drop_objs]
seen_keys: set[tuple] = set()
rewritten: list[dict] = []
for e in all_edges:
src = e.get("source")
tgt = e.get("target")
if src == tgt:
continue
k = (src, tgt, e.get("relation"), e.get("context"))
if k in seen_keys:
continue
seen_keys.add(k)
rewritten.append(e)
all_edges[:] = rewritten
def _resolve_cross_file_java_imports(
per_file: list[dict],
paths: list[Path],
) -> list[dict]:
"""Two-pass Java import resolution.
Pass 1: build a global index {ClassName: [(node_id, package), ...]} across
all Java nodes (packages come from a re-parse; node metadata doesn't carry
them).
Pass 2: re-parse each Java file; for every `import a.b.C;`, resolve C against
the index, skipping candidates whose defining file declares a different
package — an external `org.springframework.stereotype.Component` must not
link to a local `com.example.model.Component` (#2504). Wildcard and stdlib
imports produce no edge.
"""
try:
import tree_sitter_java as tsjava
from tree_sitter import Language, Parser
except ImportError:
return []
language = Language(tsjava.language())
parser = Parser(language)
# Pre-pass: declared package per source_file string (and parsed trees for
# pass 2, so each file is only parsed once).
parsed: dict[str, tuple[bytes, object]] = {}
pkg_by_src: dict[str, str] = {}
for path, file_result in zip(paths, per_file):
try:
source = path.read_bytes()
tree = parser.parse(source)
except Exception:
continue
parsed[str(path)] = (source, tree)
pkg = ""
for child in tree.root_node.children:
if child.type == "package_declaration":
pkg = _read_text(child, source).strip()[len("package"):].strip().rstrip(";").strip()
break
pkg_by_src[str(path)] = pkg
for node in file_result.get("nodes", []):
src = node.get("source_file")
if src:
pkg_by_src.setdefault(src, pkg)
def _pkg_matches(imp_pkg: str, tgt_pkg: str) -> bool:
if imp_pkg == tgt_pkg:
return True
# `import p.Outer.Inner` against a nested type defined in package p:
# the leftover segments must all be type-like (uppercase-first), which
# conventional lowercase external packages can never satisfy.
if tgt_pkg:
if not imp_pkg.startswith(tgt_pkg + "."):
return False
rest = imp_pkg[len(tgt_pkg) + 1:]
else:
rest = imp_pkg
return bool(rest) and all(seg[:1].isupper() for seg in rest.split("."))
# Pass 1: class-name → (node_id, package) index (only internal,
# uppercase-starting names)
name_to_ids: dict[str, list[tuple[str, str]]] = {}
for file_result in per_file:
for node in file_result.get("nodes", []):
label = node.get("label", "")
nid = node.get("id", "")
src = node.get("source_file", "")
if not label or not nid or not src:
continue
if label.endswith(")") or label.endswith(".java"):
continue
if not label[0].isalpha() or not label[0].isupper():
continue
name_to_ids.setdefault(label, []).append((nid, pkg_by_src.get(src, "")))
# Pass 2: resolve imports to real node IDs
new_edges: list[dict] = []
seen_pairs: set[tuple[str, str]] = set()
for path in paths:
file_nid = _make_id(str(path))
entry = parsed.get(str(path))
if entry is None:
continue
source, tree = entry
def walk(n) -> None:
if n.type == "import_declaration":
raw = _read_text(n, source).strip()
body = raw[len("import"):].strip().rstrip(";").strip()
if body.startswith("static "):
body = body[len("static "):].strip()
if body.endswith(".*"):
return
parts = body.split(".")
if not parts:
return
last = parts[-1]
imp_pkg = ".".join(parts[:-1])
if last and last[0].islower() and len(parts) >= 2:
last = parts[-2]
imp_pkg = ".".join(parts[:-2])
at_line = n.start_point[0] + 1
for tgt_nid, tgt_pkg in name_to_ids.get(last, []):
if tgt_nid == file_nid:
continue
if not _pkg_matches(imp_pkg, tgt_pkg):
continue
key = (file_nid, tgt_nid)
if key in seen_pairs:
continue
seen_pairs.add(key)
new_edges.append({
"source": file_nid,
"target": tgt_nid,
"relation": "imports",
"confidence": "EXTRACTED",
"confidence_score": 1.0,
"source_file": str(path),
"source_location": f"L{at_line}",
"weight": 1.0,
})
for child in n.children:
walk(child)
walk(tree.root_node)
return new_edges
def _go_import_path_for_file(
source_file: str | Path,
root: Path,
module_cache: dict[Path, str | None] | None = None,
) -> str | None:
"""Return the canonical Go import path for a source file inside a module."""
cache = module_cache if module_cache is not None else {}
path = Path(source_file)
if not path.is_absolute():
path = root / path
try:
directory = path.resolve().parent
except OSError:
directory = path.absolute().parent
module_dir: Path | None = None
module_path: str | None = None
for candidate in (directory, *directory.parents):
if candidate in cache:
cached = cache[candidate]
if cached:
module_dir, module_path = candidate, cached
break
go_mod = candidate / "go.mod"
if not go_mod.is_file():
continue
try:
match = re.search(
r"(?m)^\s*module\s+([^\s]+)",
go_mod.read_text(encoding="utf-8"),
)
except (OSError, UnicodeError):
match = None
module_dir = candidate
module_path = match.group(1) if match else None
cache[candidate] = module_path
break
if not module_dir or not module_path:
return None
try:
relative_dir = directory.relative_to(module_dir)
except ValueError:
return None
suffix = relative_dir.as_posix()
return module_path if suffix == "." else f"{module_path}/{suffix}"
def _resolve_go_type_references(
per_file: list[dict],
paths: list[Path],
all_nodes: list[dict],
all_edges: list[dict],
root: Path,
resolution_context_nodes: list[dict] | None = None,
resolution_context_edges: list[dict] | None = None,
) -> None:
"""Resolve qualified Go types through aliases and exact module paths."""
imports_by_file: dict[str, dict[str, str]] = {}
actual_path_by_file: dict[str, Path] = {}
for path, result in zip(paths, per_file):
imports = result.get("go_imports") or {}
for node in result.get("nodes", []):
source_file = node.get("source_file")
if source_file:
imports_by_file[str(source_file)] = imports
actual_path_by_file[str(source_file)] = path
if not imports_by_file:
return
definition_nodes = all_nodes + (resolution_context_nodes or [])
definition_edges = all_edges + (resolution_context_edges or [])
contained = {edge.get("target") for edge in definition_edges
if edge.get("relation") == "contains"}
module_cache: dict[Path, str | None] = {}
fqn_to_ids: dict[str, list[str]] = {}
for node in definition_nodes:
source_file = str(node.get("source_file") or "")
label = str(node.get("label") or "")
nid = node.get("id")
if (not source_file or not label or not nid or nid not in contained
or not _is_type_like_definition(node)):
continue
actual_path = actual_path_by_file.get(source_file, Path(source_file))
package_path = _go_import_path_for_file(actual_path, root, module_cache)
if package_path:
fqn_to_ids.setdefault(f"{package_path}.{label}", []).append(nid)
qualified_stubs = {
node["id"]: str(node.get("label") or "")
for node in all_nodes
if node.get("id") and not node.get("source_file")
and "." in str(node.get("label") or "")
}
if not qualified_stubs:
return
node_ids = {node.get("id") for node in all_nodes if node.get("id")}
external_stub_ids: dict[str, str] = {}
new_nodes: list[dict] = []
def external_stub(fqn: str) -> str:
existing = external_stub_ids.get(fqn)
if existing:
return existing
nid = _make_id("go", "type", fqn)
if nid not in node_ids:
new_nodes.append({
"id": nid,
"label": fqn,
"file_type": "code",
"source_file": "",
"source_location": "",
})
node_ids.add(nid)
external_stub_ids[fqn] = nid
return nid
repointed_from: set[str] = set()
for edge in all_edges:
if edge.get("relation") not in {"references", "embeds"}:
continue
target = edge.get("target")
qualified = qualified_stubs.get(target)
if not qualified:
continue
alias, _, type_name = qualified.rpartition(".")
import_path = imports_by_file.get(
str(edge.get("source_file") or ""), {}
).get(alias)
if not import_path or not type_name:
continue
fqn = f"{import_path}.{type_name}"
candidates = fqn_to_ids.get(fqn, [])
edge["target"] = candidates[0] if len(candidates) == 1 else external_stub(fqn)
repointed_from.add(str(target))
if new_nodes:
all_nodes.extend(new_nodes)
if not repointed_from:
return
referenced = {endpoint for edge in all_edges
for endpoint in (edge.get("source"), edge.get("target"))}
all_nodes[:] = [
node for node in all_nodes
if node.get("id") not in repointed_from or node.get("id") in referenced
]
def _resolve_java_type_references(
per_file: list[dict],
paths: list[Path],
all_nodes: list[dict],
all_edges: list[dict],
) -> None:
"""Re-point dangling Java ``implements``/``inherits`` edges to the real
definition, using the referencing file's ``import`` statements (+ package)
for exact disambiguation.
Cross-file type references resolve by bare name and fall back to a no-source
"shadow" stub. ``_rewire_unique_stub_nodes`` repairs that only when the name
is globally unique; when two packages define a same-named type it bails, so
the ``implements`` edge stays stuck on the shadow node and the real interface
is wrongly isolated (#1318). An ``import com.a.handler.AIResponseHandler``
names the exact package, so it disambiguates where bare-name matching cannot.
Mutates ``all_nodes``/``all_edges`` in place. Runs after id-disambiguation so
target ids are final, but BEFORE ``_rewire_unique_stub_nodes`` (#2504): the
rewire itself manufactures a false merge when the bare stub for an EXTERNAL
import (``org.springframework.stereotype.Component``) collapses onto the
only internal class with that simple name. References proven external by an
explicit import are re-pointed to an FQN-labeled sourceless stub the
bare-label rewire cannot collapse; references with no import/package facts
are left untouched so the legacy unique-label rewire keeps handling plain
same-package/default-package corpora (mirrors the PHP #1923 fix).
"""
try:
import tree_sitter_java as tsjava
from tree_sitter import Language, Parser
except ImportError:
return
language = Language(tsjava.language())
parser = Parser(language)
# package + simple-name->FQN imports, keyed by the source_file string the
# file's own nodes use (so it matches edge/node source_file exactly).
pkg_by_file: dict[str, str] = {}
imports_by_file: dict[str, dict[str, str]] = {}
for path, result in zip(paths, per_file):
srcs = {n.get("source_file") for n in result.get("nodes", []) if n.get("source_file")}
if not srcs:
continue
try:
source = path.read_bytes()
tree = parser.parse(source)
except Exception:
continue
pkg = ""
imps: dict[str, str] = {}
def walk(n) -> None:
nonlocal pkg
if n.type == "package_declaration":
pkg = _read_text(n, source).strip()[len("package"):].strip().rstrip(";").strip()
elif n.type == "import_declaration":
body = _read_text(n, source).strip()[len("import"):].strip().rstrip(";").strip()
if body.startswith("static "):
body = body[len("static "):].strip()
if body.endswith(".*") or "." not in body:
return
simple = body.split(".")[-1]
if simple and simple[0].isupper():
imps[simple] = body
for child in n.children:
walk(child)
walk(tree.root_node)
for s in srcs:
pkg_by_file[s] = pkg
imports_by_file[s] = imps
# FQN (package.Class or package.Outer.Inner) -> definition node id, for
# type-like defs with a source. Nested declarations need their containing
# type path because qualified annotation references preserve it.
fqn_to_id: dict[str, str] = {}
node_by_id = {
node.get("id"): node for node in all_nodes if node.get("id")
}
type_parent_by_id: dict[str, str] = {}
for edge in all_edges:
if edge.get("relation") != "contains":
continue
child = node_by_id.get(edge.get("target"))
parent = node_by_id.get(edge.get("source"))
if not child or not parent:
continue
child_src = child.get("source_file", "")
parent_label = parent.get("label", "")
if (
child_src
and parent.get("source_file") == child_src
and parent_label[:1].isupper()
and not parent_label.endswith(".java")
):
type_parent_by_id[child["id"]] = parent["id"]
for node in all_nodes:
label = node.get("label", "")
src = node.get("source_file", "")
nid = node.get("id", "")
if not (label and src and nid) or src not in pkg_by_file:
continue
if not label[:1].isupper() or label.endswith(")") or label.endswith(".java"):
continue
pkg = pkg_by_file[src]
fqn_to_id.setdefault(f"{pkg}.{label}" if pkg else label, nid)
type_path = [label]
seen = {nid}
parent_id = type_parent_by_id.get(nid)
while parent_id and parent_id not in seen:
seen.add(parent_id)
parent = node_by_id[parent_id]
type_path.append(parent["label"])
parent_id = type_parent_by_id.get(parent_id)
if len(type_path) > 1:
nested_name = ".".join(reversed(type_path))
fqn_to_id.setdefault(
f"{pkg}.{nested_name}" if pkg else nested_name,
nid,
)
# Shadow stubs: no source_file, type-like label. Dotted labels are included
# for qualified inline annotations (`@com.example.anno.Loggable`), which the
# engine mints with their full dotted name so a same-named local class can't
# absorb them (#2504).
stub_label: dict[str, str] = {
node["id"]: node.get("label", "")
for node in all_nodes
if node.get("id") and not node.get("source_file")
and (node.get("label", "")[:1].isupper() or "." in node.get("label", ""))
}
if not stub_label:
return
# `imports` is included so the file-level import edge that also lands on the
# shadow stub gets re-pointed too, leaving the stub unreferenced (and dropped).
# External/stdlib imports never resolve (no internal def / same-package match),
# so their edges correctly stay on their stub. `references` (field/parameter/
# return-type uses) is included so a cross-module reference to a same-named
# class doesn't dangle on a sourceless phantom node when two packages define
# the same simple name — the node itself survives with a path-scoped id, but
# the reference must point at the RIGHT one (#1744). Mirrors the C# resolver,
# whose REPOINT set already covers `references`.
REPOINT_RELATIONS = {"implements", "inherits", "extends", "imports", "references"}
node_ids = {n.get("id") for n in all_nodes if n.get("id")}
external_stub_ids: dict[str, str] = {}
new_nodes: list[dict] = []
def _external_stub(fqn: str) -> str:
nid = external_stub_ids.get(fqn)
if nid:
return nid
nid = _make_id(fqn)
if nid not in node_ids:
new_nodes.append({
"id": nid,
"label": fqn,
"file_type": "code",
"source_file": "",
"source_location": "",
})
node_ids.add(nid)
external_stub_ids[fqn] = nid
return nid
repointed_from: set[str] = set()
for edge in all_edges:
if edge.get("relation") not in REPOINT_RELATIONS:
continue
tgt = edge.get("target")
label = stub_label.get(tgt)
if not label:
continue
ref_file = edge.get("source_file", "")
if "." in label:
# FQN-labeled stub (qualified inline annotation): resolve it against
# the internal definitions; an external FQN stays parked as-is.
resolved = fqn_to_id.get(label)
if resolved and resolved != tgt:
edge["target"] = resolved
repointed_from.add(tgt)
continue
fqn = imports_by_file.get(ref_file, {}).get(label)
if fqn:
resolved = fqn_to_id.get(fqn)
if resolved is None:
# `import p.Outer.Inner`: strip trailing type-like segments to
# find the defining package of an internal nested type.
head = fqn.split(".")[:-1]
while resolved is None and head and head[-1][:1].isupper():
head.pop()
resolved = fqn_to_id.get(".".join(head + [label]))
if resolved is None:
# Explicit import with no internal definition: proven EXTERNAL.
# Park the edge on an FQN-labeled stub the bare-name rewire
# cannot collapse onto a same-named local class (#2504 — this
# is the Java counterpart of the PHP #1923 fix).
edge["target"] = _external_stub(fqn)
repointed_from.add(tgt)
continue
else: # same-package reference (no explicit import)
pkg = pkg_by_file.get(ref_file, "")
resolved = fqn_to_id.get(f"{pkg}.{label}" if pkg else label)
if resolved and resolved != tgt:
edge["target"] = resolved
repointed_from.add(tgt)
if new_nodes:
all_nodes.extend(new_nodes)
# Bare imported and inline-qualified annotation references can start on
# different stubs, then converge on one source-backed node above. Collapse
# only indistinguishable Java attribute-reference facts after that rewire.
seen_attribute_refs: set[tuple] = set()
deduped_edges: list[dict] = []
for edge in all_edges:
if (
edge.get("relation") == "references"
and edge.get("context") == "attribute"
and edge.get("source_file", "") in pkg_by_file
):
key = (
edge.get("source"),
edge.get("target"),
edge.get("relation"),
edge.get("context"),
edge.get("source_file"),
edge.get("source_location"),
)
if key in seen_attribute_refs:
continue
seen_attribute_refs.add(key)
deduped_edges.append(edge)
all_edges[:] = deduped_edges
if not repointed_from:
return
# Drop shadow stubs that no edge references anymore.
still_referenced: set[str] = set()
for edge in all_edges:
still_referenced.add(edge.get("source"))
still_referenced.add(edge.get("target"))
all_nodes[:] = [
node for node in all_nodes
if node.get("id") not in repointed_from or node.get("id") in still_referenced
]
_PHP_SUPERTYPE_RELATIONS = ("inherits", "implements", "mixes_in")
_PHP_REPOINT_RELATIONS = frozenset({"inherits", "implements", "mixes_in", "imports", "references"})
def _php_fqn_from_raw(raw: str, ns: str, uses: dict[str, str]) -> str:
"""Resolve a raw (possibly qualified) PHP class reference to an FQN.
PHP name-resolution for class names:
\\A\\B -> absolute: A\\B
A\\B -> first segment through the `use` map (group-prefix semantics),
else relative to the current namespace
B -> `use` map, else current namespace (class names do NOT fall
back to the global namespace)
"""
raw = raw.strip()
if raw.startswith("\\"):
return raw[1:]
if "\\" in raw:
first, rest = raw.split("\\", 1)
mapped = uses.get(first.lower())
if mapped:
return f"{mapped}\\{rest}"
return f"{ns}\\{raw}" if ns else raw
mapped = uses.get(raw.lower())
if mapped:
return mapped
return f"{ns}\\{raw}" if ns else raw
def _resolve_php_type_references(
per_file: list[dict],
paths: list[Path],
all_nodes: list[dict],
all_edges: list[dict],
) -> None:
"""Disambiguate PHP inherits/implements/mixes_in/imports/references targets
using each file's ``namespace`` declaration and ``use`` imports (#1923).
Mirrors ``_resolve_java_type_references`` (a re-parse pass), but MUST run
BEFORE ``_rewire_unique_stub_nodes``: the false edge is manufactured by the
rewire itself — a bare ``Page`` stub collapses onto the only internal class
labeled ``Page`` even though the referencing file ``use``d a different
namespace (``Filament\\Pages\\Page`` vs ``App\\Models\\Page``). References
proven external by a ``use`` FQN or a qualified name are re-pointed to an
FQN-labeled sourceless stub, which the bare-label rewire cannot collapse.
References with no namespace facts are left untouched so the unique-label
rewire keeps handling plain (non-namespaced) PHP as before.
"""
try:
import tree_sitter_php as tsphp
from tree_sitter import Language, Parser
except ImportError:
return
lang_fn = getattr(tsphp, "language_php", None) or getattr(tsphp, "language", None)
if lang_fn is None:
return
language = Language(lang_fn())
parser = Parser(language)
ns_by_file: dict[str, str] = {}
uses_by_file: dict[str, dict[str, str]] = {} # lower alias -> FQN
raw_by_file: dict[str, dict[tuple[str, str], str | None]] = {} # (relation, lower bare) -> raw | None(ambiguous)
for path, result in zip(paths, per_file):
srcs = {n.get("source_file") for n in result.get("nodes", []) if n.get("source_file")}
if not srcs:
continue
try:
source = path.read_bytes()
tree = parser.parse(source)
except Exception:
continue
namespaces: list[str] = []
uses: dict[str, str] = {}
raws: dict[tuple[str, str], str | None] = {}
def _record_raw(relation: str, raw: str) -> None:
bare = raw.rsplit("\\", 1)[-1].strip().lower()
if not bare:
return
key = (relation, bare)
if key in raws and raws[key] != raw:
raws[key] = None # e.g. `implements A\I, B\I` — never guess
else:
raws.setdefault(key, raw)
def _record_use_clause(clause, prefix: str) -> None:
target = None
alias = None
saw_as = False
for c in clause.children:
if c.type in ("function", "const"):
return # not a class import
if c.type == "as":
saw_as = True
elif c.type in ("qualified_name", "name"):
if saw_as:
alias = _read_text(c, source)
elif target is None:
target = _read_text(c, source)
if not target:
return
fqn = (f"{prefix}\\{target}" if prefix else target).lstrip("\\")
key = (alias or fqn.rsplit("\\", 1)[-1]).strip().lower()
if key:
uses.setdefault(key, fqn)
def walk(n) -> None:
t = n.type
if t == "namespace_definition":
for c in n.children:
if c.type == "namespace_name":
namespaces.append(_read_text(c, source))
break
elif t == "namespace_use_declaration":
prefix = ""
group = None
for c in n.children:
if c.type == "namespace_name":
prefix = _read_text(c, source) # group-use prefix
elif c.type == "namespace_use_group":
group = c
elif c.type == "namespace_use_clause":
_record_use_clause(c, "")
if group is not None:
for c in group.children:
if c.type == "namespace_use_clause":
_record_use_clause(c, prefix)
return
elif t == "class_declaration":
for child in n.children:
if child.type == "base_clause":
for sub in child.children:
if sub.type in ("name", "qualified_name"):
_record_raw("inherits", _read_text(sub, source))
elif child.type == "class_interface_clause":
for sub in child.children:
if sub.type in ("name", "qualified_name"):
_record_raw("implements", _read_text(sub, source))
elif child.type == "declaration_list":
for member in child.children:
if member.type != "use_declaration":
continue
for sub in member.children:
if sub.type in ("name", "qualified_name"):
_record_raw("mixes_in", _read_text(sub, source))
for child in n.children:
walk(child)
walk(tree.root_node)
if len(set(namespaces)) > 1:
continue # multi-namespace file (PSR-1 violation): keep legacy behavior
ns = namespaces[0] if namespaces else ""
for s in srcs:
ns_by_file[s] = ns
uses_by_file[s] = uses
raw_by_file[s] = raws
if not ns_by_file:
return
# lower FQN -> definition node id (PHP class names are case-insensitive).
fqn_to_id: dict[str, str] = {}
for node in all_nodes:
label = node.get("label", "")
src = node.get("source_file", "")
nid = node.get("id", "")
if not (label and src and nid) or src not in ns_by_file:
continue
if label.endswith(")") or "." in label: # methods / file nodes
continue
ns = ns_by_file[src]
fqn = f"{ns}\\{label}" if ns else label
fqn_to_id.setdefault(fqn.lower(), nid)
node_ids = {n.get("id") for n in all_nodes if n.get("id")}
stub_label: dict[str, str] = {
n["id"]: n.get("label", "")
for n in all_nodes
if n.get("id") and not n.get("source_file") and n.get("label")
}
external_stub_ids: dict[str, str] = {}
new_nodes: list[dict] = []
def _external_stub(fqn: str) -> str:
key = fqn.lower()
nid = external_stub_ids.get(key)
if nid:
return nid
nid = _make_id(fqn)
if nid not in node_ids:
new_nodes.append({
"id": nid,
"label": fqn,
"file_type": "code",
"source_file": "",
"source_location": "",
})
node_ids.add(nid)
external_stub_ids[key] = nid
return nid
repointed_from: set[str] = set()
for edge in all_edges:
relation = edge.get("relation")
if relation not in _PHP_REPOINT_RELATIONS:
continue
ref_file = edge.get("source_file", "")
if ref_file not in ns_by_file:
continue
tgt = edge.get("target")
label = stub_label.get(tgt)
uses = uses_by_file.get(ref_file, {})
if not label and relation == "imports":
label = next((alias for alias in uses if _make_id(alias) == tgt), "")
if not label:
continue
bare = label.strip().lower()
ns = ns_by_file[ref_file]
raw = None
if relation in _PHP_SUPERTYPE_RELATIONS:
raw = raw_by_file.get(ref_file, {}).get((relation, bare))
explicit = False
if raw and "\\" in raw:
fqn = _php_fqn_from_raw(raw, ns, uses)
explicit = True
elif bare in uses:
fqn = uses[bare]
explicit = True
elif ns:
fqn = f"{ns}\\{label}"
else:
continue # no namespace facts: legacy unique-label rewire applies
resolved = fqn_to_id.get(fqn.lower())
if resolved and resolved != tgt:
edge["target"] = resolved
repointed_from.add(tgt)
elif explicit and resolved is None:
# Proven external: park the edge on an FQN-labeled stub the
# bare-name rewire cannot collapse (this is the #1923 fix).
edge["target"] = _external_stub(fqn)
repointed_from.add(tgt)
# non-explicit miss: leave the bare stub for the legacy rewire
if new_nodes:
all_nodes.extend(new_nodes)
if not repointed_from:
return
still_referenced: set[str] = set()
for edge in all_edges:
still_referenced.add(edge.get("source"))
still_referenced.add(edge.get("target"))
all_nodes[:] = [
n for n in all_nodes
if n.get("id") not in repointed_from or n.get("id") in still_referenced
]
_pascal_unit_cache: dict[str, dict[str, str]] = {}
_pascal_class_stem_cache: dict[str, dict[str, str]] = {} # root_key → {stem_lower: _file_stem}
def _pascal_project_root(from_path: Path) -> Path:
"""Return the highest ancestor directory that looks like a Pascal project root.
Walks up the directory tree and tracks the topmost directory that:
- is NOT a filesystem root (e.g. D:/, C:/, /)
- has at least 2 .pas files OR at least 1 .dpr file as direct children
The minimum-2 threshold avoids treating a level as the root just because a
single stray .pas file was copied there. The filesystem-root exclusion
prevents overshoot on drives that have a stray file directly at D:/.
Falls back to from_path.parent if nothing better is found.
"""
best = from_path.parent
current = from_path.parent
for _ in range(12):
if len(current.parts) <= 1:
break # never use a filesystem root (D:/, C:/, /)
pas_count = sum(1 for _ in current.glob("*.pas"))
dpr_count = sum(1 for _ in current.glob("*.dpr"))
if pas_count >= 2 or dpr_count >= 1:
best = current
parent = current.parent
if parent == current:
break
current = parent
return best
def _pascal_resolve_unit(from_path: Path, unit_name: str) -> str:
"""Resolve a Pascal unit name to the graphify node ID of its source file.
Scans all Pascal files under the project root (the highest ancestor that
directly contains .pas/.dpr files) and returns _make_id(str(matched_path)).
Result is cached per project root so the rglob runs at most once per
project. Falls back to _make_id(unit_name) for units not found on disk
(e.g. standard RTL units like SysUtils, Windows).
"""
root = _pascal_project_root(from_path)
root_key = str(root)
if root_key not in _pascal_unit_cache:
unit_map: dict[str, str] = {}
for ext in (".pas", ".pp", ".dpr", ".dpk", ".inc"):
for f in root.rglob("*" + ext):
unit_map[f.stem.lower()] = _make_id(str(f))
_pascal_unit_cache[root_key] = unit_map
return _pascal_unit_cache[root_key].get(unit_name.lower(), _make_id(unit_name))
def _pascal_resolve_class(from_path: Path, class_name: str) -> str | None:
"""Resolve a Pascal class/interface name to the node ID of its defining file's class node.
Pascal convention: TFooBar is defined in FooBar.pas, IFooBar in FooBar.pas.
Strips the leading T/I prefix, finds the file, and returns
_make_id(_file_stem(found_file), class_name).
Returns None when no matching file is found on disk (RTL, stdlib, or
unconventionally-named class — caller should create a stub node).
"""
prefix = class_name[:1]
unit_name = class_name[1:] if prefix in ("T", "I") else class_name
root = _pascal_project_root(from_path)
root_key = str(root)
if root_key not in _pascal_class_stem_cache:
stem_map: dict[str, str] = {}
for ext in (".pas", ".pp", ".dpr", ".dpk"):
for f in root.rglob("*" + ext):
stem_map[f.stem.lower()] = _file_stem(f)
_pascal_class_stem_cache[root_key] = stem_map
file_stem = _pascal_class_stem_cache[root_key].get(unit_name.lower())
if file_stem:
return _make_id(file_stem, class_name)
return None
Reference in New Issue View Git Blame Copy Permalink