diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d818561b..e547f2c3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ Full release notes with details on each version: [GitHub Releases](https://github.com/safishamsi/graphify/releases) +## Unreleased + +- Feature: `detect()` enumerates large trees via `git ls-files`, Google `repo` project lists, and nested git worktrees/submodules, so gitignored directories are not walked. +- Feature: AST extract, JS/Python symbol resolution, and cache probes share `graphify.parallel` process/thread pools (`GRAPHIFY_MAX_WORKERS`). +- Feature: `graphify extract` is quiet by default; pass `--verbose` (or `GRAPHIFY_VERBOSE=1`) for detect/AST/resolving progress. `--timing` is unchanged. Symbol resolution now reports JS/TS facts, Python facts, and applying edges so that stage is not a black box. +- Fix: merge-driver git config round-trips a quoted Windows interpreter path on POSIX git (#2166). + ## 0.9.53 (2026-08-30) - Fix: a batch of cross-language inheritance-edge corrections (thanks @Synvoya): JavaScript `class X extends Y` now emits an `inherits` edge (#1790); PHP interfaces, enums, and traits are captured as class-like nodes with their heritage (#1791); Scala `trait` declarations become class-like nodes (#1792) and qualified `extends`/`with` bases resolve to the tail type (#1794); a qualified Kotlin supertype resolves to its tail type instead of the package head (#1793); a C# interface extending an interface is classified as `inherits`, not `implements` (#1817); and a Go interface type-set constraint no longer emits a spurious `embeds` edge (#1818). diff --git a/README.md b/README.md index d83ea76c0..3fb20b12f 100644 --- a/README.md +++ b/README.md @@ -773,6 +773,7 @@ graphify extract ./docs --google-workspace # export .gdoc/.gsheet/.gslides v graphify extract ./src --no-gitignore # include git-ignored source; still honor .graphifyignore graphify extract ./docs --mode deep # richer semantic extraction via extended system prompt graphify extract ./docs --no-cluster # raw extraction only, skip clustering +graphify extract ./docs --verbose # progress on detect, AST, and symbol resolution graphify extract ./docs --timing # print per-stage wall-clock timings to stderr (also works on cluster-only) graphify extract ./docs --force # overwrite graph.json even if new graph has fewer nodes (use after refactors or to clear ghost duplicates) graphify extract ./docs --dedup-llm # LLM tiebreaker for ambiguous entity pairs (uses same API key) diff --git a/graphify/__main__.py b/graphify/__main__.py index 4a68e7240..71cf53d8a 100644 --- a/graphify/__main__.py +++ b/graphify/__main__.py @@ -639,6 +639,8 @@ def _run_cli() -> None: print(" --no-gitignore ignore .gitignore and .git/info/exclude (prioritizes .graphifyignore)") print(" --no-cluster skip clustering, write raw extraction only") print(" --code-only index code (local AST, no API key) and skip doc/paper/image files") + print(" --verbose progress on detect, AST, and symbol resolution (quiet by default)") + print(" --timing per-stage wall-clock timings on stderr") print(" --postgres DSN extract schema from a live PostgreSQL database") print(" maps tables, views, functions + FK relationships;") print(" column-level detail is not represented in the graph") diff --git a/graphify/cache.py b/graphify/cache.py index 622ba6cff..8aaf40888 100644 --- a/graphify/cache.py +++ b/graphify/cache.py @@ -7,6 +7,7 @@ import os import re import tempfile +import threading import time import warnings from collections.abc import Callable, Iterable @@ -198,6 +199,9 @@ def _body_content(content: bytes) -> bytes: # `graphify extract --force` / `graphify update --force` (or GRAPHIFY_FORCE=1) # skip the cache reads and re-dispatch everything when needed (#1894). _stat_index: dict[str, dict] = {} +# Detect word-count and extract cache probes run in a thread pool; every +# read/write of this dict (and the dirty flag) goes through this lock. +_stat_index_lock = threading.RLock() _stat_index_root: Path | None = None # Key anchor for the ON-DISK index (#2199): the first caller's key-root, i.e. # the corpus. Distinct from _stat_index_root, which is the cache-FILE location @@ -329,89 +333,103 @@ def _ensure_stat_index(root: Path, cache_root: "Path | None" = None) -> None: global _stat_index, _stat_index_root, _stat_index_anchor, _stat_index_dirty if _stat_index_root is not None: return - # _stat_index_root determines the cache FILE location, so honoring an - # explicit cache_root keeps detect()'s word-count cache under the requested - # --out dir instead of polluting the scanned corpus with a stray - # graphify-out/ (#1747). _stat_index_anchor is the separate KEY anchor: - # in-memory keys stay absolute, but the on-disk index stores in-anchor keys - # relative so a moved/cloned corpus still hits (#2199) — same load/save - # re-anchoring the detect manifest uses. - _stat_index_root = Path(cache_root if cache_root is not None else root).resolve() - _stat_index_anchor = Path(root).resolve() - p = _stat_index_file(_stat_index_root) - _stat_index = {} - if p.exists(): - try: - raw = json.loads(p.read_text(encoding="utf-8")) - if isinstance(raw, dict): - for k, v in raw.items(): - if not isinstance(k, str): - continue - if Path(k).is_absolute(): - # Legacy/out-of-anchor key: pass through, but never - # clobber a re-anchored relative (new-format) entry - # that resolved to the same absolute path. - _stat_index.setdefault(k, v) - else: - _stat_index[_stat_key_to_absolute(k, _stat_index_anchor)] = v - except (json.JSONDecodeError, OSError): - _stat_index = {} - atexit.register(_flush_stat_index) + with _stat_index_lock: + if _stat_index_root is not None: + return + # _stat_index_root determines the cache FILE location, so honoring an + # explicit cache_root keeps detect()'s word-count cache under the requested + # --out dir instead of polluting the scanned corpus with a stray + # graphify-out/ (#1747). _stat_index_anchor is the separate KEY anchor: + # in-memory keys stay absolute, but the on-disk index stores in-anchor keys + # relative so a moved/cloned corpus still hits (#2199) — same load/save + # re-anchoring the detect manifest uses. + # + # Publish ``_stat_index_root`` last: the unlocked fast-path above treats + # a non-None root as "index is ready". Setting it first raced a second + # thread into an empty dict while this thread was still loading. + location = Path(cache_root if cache_root is not None else root).resolve() + anchor = Path(root).resolve() + p = _stat_index_file(location) + index: dict[str, dict] = {} + if p.exists(): + try: + raw = json.loads(p.read_text(encoding="utf-8")) + if isinstance(raw, dict): + for k, v in raw.items(): + if not isinstance(k, str): + continue + if Path(k).is_absolute(): + # Legacy/out-of-anchor key: pass through, but never + # clobber a re-anchored relative (new-format) entry + # that resolved to the same absolute path. + index.setdefault(k, v) + else: + index[_stat_key_to_absolute(k, anchor)] = v + except (json.JSONDecodeError, OSError): + index = {} + _stat_index = index + _stat_index_anchor = anchor + atexit.register(_flush_stat_index) + _stat_index_root = location def _flush_stat_index() -> None: global _stat_index_dirty, _stat_index_root - if not _stat_index_dirty or _stat_index_root is None: - return - p = _stat_index_file(_stat_index_root) - # Build the on-disk form (#2199): prune entries whose file is gone (the - # index otherwise grows without bound), then store in-anchor keys as - # forward-slash relative paths so the index survives a corpus move/clone. - # Out-of-anchor keys stay absolute (same rule as the detect manifest); a - # reader tells the formats apart by absoluteness, so no version marker is - # needed. In-memory keys are untouched — only the serialization changes. - on_disk: dict[str, dict] = {} - for k, v in _stat_index.items(): - try: - if not os.path.exists(k): + with _stat_index_lock: + if not _stat_index_dirty or _stat_index_root is None: + return + p = _stat_index_file(_stat_index_root) + # Snapshot under the lock so a concurrent hash/word-count cannot mutate + # the dict we are serializing. The file write stays inside the lock: + # atexit vs a late worker is rare, and a torn write is worse. + # Build the on-disk form (#2199): prune entries whose file is gone (the + # index otherwise grows without bound), then store in-anchor keys as + # forward-slash relative paths so the index survives a corpus move/clone. + # Out-of-anchor keys stay absolute (same rule as the detect manifest); a + # reader tells the formats apart by absoluteness, so no version marker is + # needed. In-memory keys are untouched — only the serialization changes. + on_disk: dict[str, dict] = {} + for k, v in _stat_index.items(): + try: + if not os.path.exists(k): + continue + except OSError: continue + dk = _stat_key_to_relative(k, _stat_index_anchor) if _stat_index_anchor is not None else k + on_disk[dk] = v + # Never resurrect a corpus that was deleted while graphify was running + # (#2974): a hook-launched `graphify update . &` in a short-lived worktree + # outlives `git worktree remove`, and an unconditional `mkdir -p` here + # rebuilt the dead path as a husk holding nothing but this index. The + # index is a pure optimisation, so when its root is gone it is simply not + # written. Creating graphify-out/cache/ under a root that still exists is + # unchanged (a first run writes the index before anything else does). + try: + if not _stat_index_root.is_dir(): + _stat_index_dirty = False + return except OSError: - continue - dk = _stat_key_to_relative(k, _stat_index_anchor) if _stat_index_anchor is not None else k - on_disk[dk] = v - # Never resurrect a corpus that was deleted while graphify was running - # (#2974): a hook-launched `graphify update . &` in a short-lived worktree - # outlives `git worktree remove`, and an unconditional `mkdir -p` here - # rebuilt the dead path as a husk holding nothing but this index. The - # index is a pure optimisation, so when its root is gone it is simply not - # written. Creating graphify-out/cache/ under a root that still exists is - # unchanged (a first run writes the index before anything else does). - try: - if not _stat_index_root.is_dir(): _stat_index_dirty = False return - except OSError: - _stat_index_dirty = False - return - try: - p.parent.mkdir(parents=True, exist_ok=True) - fd, tmp = tempfile.mkstemp(dir=p.parent, prefix="stat-index.", suffix=".tmp") try: - os.write(fd, json.dumps(on_disk, separators=(",", ":")).encode()) - os.close(fd) - os.replace(tmp, p) - except Exception: + p.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=p.parent, prefix="stat-index.", suffix=".tmp") try: + os.write(fd, json.dumps(on_disk, separators=(",", ":")).encode()) os.close(fd) - except OSError: - pass - try: - os.unlink(tmp) - except OSError: - pass - except OSError: - pass - _stat_index_dirty = False + os.replace(tmp, p) + except Exception: + try: + os.close(fd) + except OSError: + pass + try: + os.unlink(tmp) + except OSError: + pass + except OSError: + pass + _stat_index_dirty = False def _normalize_path(path: Path) -> Path: @@ -495,14 +513,15 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No st: "os.stat_result | None" = None try: st = p.stat() - if _stat_sig_fresh(_stat_index.get(abs_key), st): - hashes = _stat_index[abs_key].get("hashes") - if isinstance(hashes, dict): - cached = hashes.get(salt) - if isinstance(cached, str): - return cached - # Legacy single-digest entries ("hash") don't record which salt - # produced them, so they are never trusted (#1989) — recompute once. + with _stat_index_lock: + if _stat_sig_fresh(_stat_index.get(abs_key), st): + hashes = _stat_index[abs_key].get("hashes") + if isinstance(hashes, dict): + cached = hashes.get(salt) + if isinstance(cached, str): + return cached + # Legacy single-digest entries ("hash") don't record which salt + # produced them, so they are never trusted (#1989) — recompute once. except OSError: pass @@ -518,14 +537,15 @@ def file_hash(path: Path, root: Path = Path("."), cache_root: "Path | None" = No digest = h.hexdigest() if st is not None: - entry = _stat_entry_for(abs_key, st, observed_at_ns) - hashes = entry.get("hashes") - if not isinstance(hashes, dict): - hashes = {} - entry["hashes"] = hashes - hashes[salt] = digest # preserve a co-located word_count / other salts - entry.pop("hash", None) # retire the un-salted legacy digest - _stat_index_dirty = True + with _stat_index_lock: + entry = _stat_entry_for(abs_key, st, observed_at_ns) + hashes = entry.get("hashes") + if not isinstance(hashes, dict): + hashes = {} + entry["hashes"] = hashes + hashes[salt] = digest # preserve a co-located word_count / other salts + entry.pop("hash", None) # retire the un-salted legacy digest + _stat_index_dirty = True return digest @@ -551,9 +571,10 @@ def cached_word_count(path: Path, root: Path, compute, cache_root: "Path | None" st: "os.stat_result | None" = None try: st = p.stat() - entry = _stat_index.get(abs_key) - if _stat_sig_fresh(entry, st) and "word_count" in entry: - return entry["word_count"] + with _stat_index_lock: + entry = _stat_index.get(abs_key) + if _stat_sig_fresh(entry, st) and "word_count" in entry: + return entry["word_count"] except OSError: pass @@ -563,8 +584,9 @@ def cached_word_count(path: Path, root: Path, compute, cache_root: "Path | None" wc = compute(Path(path)) if st is not None: - _stat_entry_for(abs_key, st, observed_at_ns)["word_count"] = wc - _stat_index_dirty = True + with _stat_index_lock: + _stat_entry_for(abs_key, st, observed_at_ns)["word_count"] = wc + _stat_index_dirty = True return wc diff --git a/graphify/cli.py b/graphify/cli.py index 6642f57d5..a6ed2df93 100644 --- a/graphify/cli.py +++ b/graphify/cli.py @@ -649,12 +649,26 @@ def __init__(self, enabled: bool) -> None: def mark(self, stage: str) -> None: now = self._now() if self.enabled: - print(f"[graphify timing] {stage}: {now - self._last:.1f}s", file=sys.stderr) + print( + f"[graphify timing] {stage}: {now - self._last:.1f}s", + file=sys.stderr, flush=True, + ) self._last = now def total(self) -> None: if self.enabled: - print(f"[graphify timing] total: {self._now() - self.start:.1f}s", file=sys.stderr) + print( + f"[graphify timing] total: {self._now() - self.start:.1f}s", + file=sys.stderr, flush=True, + ) + + +def _extract_progress(msg: str) -> None: + """Stage progress; printed only with ``--verbose`` / ``GRAPHIFY_VERBOSE``.""" + from graphify.progress import vprint + vprint(msg, file=sys.stdout, prefix="[graphify extract]") + + def _enforce_graph_size_cap_or_exit(gp: Path) -> None: """Reject oversized graph files before parsing (CLI exit-on-fail flavor). @@ -3148,7 +3162,7 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": "[--model M] [--mode deep] [--out DIR|--output DIR] [--google-workspace] [--no-cluster] " "[--no-gitignore] [--code-only] [--no-dedup] " "[--max-workers N] [--token-budget N] [--max-concurrency N] " - "[--api-timeout S] [--postgres DSN] [--cargo] [--allow-partial] [--timing]", + "[--api-timeout S] [--postgres DSN] [--cargo] [--allow-partial] [--timing] [--verbose]", file=sys.stderr, ) sys.exit(1) @@ -3194,6 +3208,7 @@ def _to_simple(g: "_nx.Graph") -> "_nx.Graph": cli_exclude_hubs: float | None = None cli_excludes: list[str] = [] cli_timing: bool = False + cli_verbose: bool = False # --force parity with `graphify update`: the flag or GRAPHIFY_FORCE=1 # disables the incremental gate and skips semantic-cache reads (#1894). force = os.environ.get("GRAPHIFY_FORCE", "").lower() in ("1", "true", "yes") @@ -3300,6 +3315,8 @@ def _parse_float(name: str, raw: str) -> float: cli_allow_partial = True; i += 1 elif a == "--timing": cli_timing = True; i += 1 + elif a == "--verbose": + cli_verbose = True; i += 1 else: i += 1 @@ -3336,6 +3353,9 @@ def _parse_float(name: str, raw: str) -> float: if cli_max_workers is not None: os.environ["GRAPHIFY_MAX_WORKERS"] = str(cli_max_workers) + from graphify.progress import set_verbose + set_verbose(True if cli_verbose else None) + # Resolve output dir. The user-facing contract is "/graphify-out/" # so a fresh checkout writes graphify-out/ at the project root, matching # the skill.md pipeline. @@ -3429,6 +3449,7 @@ def _parse_float(name: str, raw: str) -> float: google_workspace=google_workspace or None, extra_excludes=_effective_excludes or None, gitignore=_effective_gitignore, + code_only=code_only, ) files_by_type = detection.get("files", {}) new_by_type = detection.get("new_files", {}) @@ -3496,13 +3517,14 @@ def _parse_float(name: str, raw: str) -> float: if _p in _healed_sem_set: image_files.append(Path(_p)) else: - print(f"[graphify extract] scanning {target}") + _extract_progress(f"scanning {target}") detection = _detect( target, google_workspace=google_workspace or None, extra_excludes=_effective_excludes or None, cache_root=out_root, gitignore=_effective_gitignore, + code_only=code_only, ) files_by_type = detection.get("files", {}) code_files = [Path(p) for p in files_by_type.get("code", [])] @@ -3564,15 +3586,15 @@ def _parse_float(name: str, raw: str) -> float: # (#1908): they still exist on disk, the scan just stopped # covering them (ignore rules / --exclude changed). _excl_note = f"; {len(excluded_files)} excluded" if excluded_files else "" - print( - f"[graphify extract] {len(code_files)} code, {len(doc_files)} docs, " + _extract_progress( + f"{len(code_files)} code, {len(doc_files)} docs, " f"{len(paper_files)} papers, {len(image_files)} images changed; " f"{unchanged_total} unchanged; {len(deleted_files)} deleted" f"{_excl_note}" ) else: - print( - f"[graphify extract] found {len(code_files)} code, " + _extract_progress( + f"found {len(code_files)} code, " f"{len(doc_files)} docs, {len(paper_files)} papers, " f"{len(image_files)} images" ) @@ -3585,7 +3607,8 @@ def _parse_float(name: str, raw: str) -> float: _more = f" (+{len(_unclassified) - 6} more)" if len(_unclassified) > 6 else "" print( f"[graphify extract] {len(_unclassified)} file(s) not classified " - f"(no supported extension or shebang), skipped: {_names}{_more}" + f"(no supported extension or shebang), skipped: {_names}{_more}", + flush=True, ) # Name the files dropped by the sensitive-file filter so a wrongly-flagged # source/doc is visible, not just a count (#2106). Operational skips @@ -3598,9 +3621,14 @@ def _parse_float(name: str, raw: str) -> float: _smore = f" (+{len(_sec) - 6} more)" if len(_sec) > 6 else "" print( f"[graphify extract] {len(_sec)} file(s) skipped as potentially sensitive " - f"(rename or move if wrongly flagged): {_snames}{_smore}" + f"(rename or move if wrongly flagged): {_snames}{_smore}", + flush=True, ) stages.mark("detect") + _extract_progress( + f"detect done — next: AST on {len(code_files)} code file(s)" + + (f", semantic on {len(semantic_files)}" if semantic_files else "") + ) # Resolve the LLM backend only now that we know whether the corpus # needs one. A code-only corpus is pure local AST and must not require @@ -3709,6 +3737,7 @@ def _parse_float(name: str, raw: str) -> float: # the issue #698 case — skip cleanly instead of crashing inside extract(). ast_result: dict = {"nodes": [], "edges": [], "input_tokens": 0, "output_tokens": 0} if code_files: + _extract_progress("loading AST extractors...") from graphify.extract import extract as _ast_extract # Anchor the cache at the output root, not the scanned project: # with --out, a /graphify-out/cache/ would leak a @@ -3800,7 +3829,9 @@ def _ctx_identity(source_file) -> str | None: ast_kwargs["resolution_context_nodes"] = _ctx_nodes if _ctx_edges: ast_kwargs["resolution_context_edges"] = _ctx_edges - print(f"[graphify extract] AST extraction on {len(code_files)} code files...") + _extract_progress( + f"AST extraction on {len(code_files)} code files..." + ) try: ast_result = _ast_extract(code_files, **ast_kwargs) except Exception as exc: @@ -3815,6 +3846,11 @@ def _ctx_identity(source_file) -> str | None: ast_result = {"nodes": [], "edges": [], "input_tokens": 0, "output_tokens": 0} _extraction_incomplete = True # the whole AST pass was lost stages.mark("AST extract") + if code_files: + _extract_progress( + f"AST done: {len(ast_result.get('nodes', []))} nodes, " + f"{len(ast_result.get('edges', []))} edges" + ) # Semantic extraction on docs/papers/images. Check cache first. from graphify.cache import ( @@ -3862,10 +3898,14 @@ def _ctx_identity(source_file) -> str | None: sem_result["edges"].extend(cached_edges) sem_result["hyperedges"].extend(cached_hyperedges) if sem_cache_hits: - print(f"[graphify extract] semantic cache: {sem_cache_hits} hit / {sem_cache_misses} miss") + _extract_progress( + f"semantic cache: {sem_cache_hits} hit / {sem_cache_misses} miss" + ) if uncached_paths: - print(f"[graphify extract] semantic extraction on {len(uncached_paths)} files via {backend}...") + _extract_progress( + f"semantic extraction on {len(uncached_paths)} files via {backend}..." + ) corpus_kwargs: dict = { "backend": backend, "model": model, @@ -3887,10 +3927,7 @@ def _ctx_identity(source_file) -> str | None: def _progress(idx: int, total: int, _result: dict) -> None: _chunk_stats["total"] = total _chunk_stats["succeeded"] += 1 - print( - f"[graphify extract] chunk {idx + 1}/{total} done", - flush=True, - ) + _extract_progress(f"chunk {idx + 1}/{total} done") corpus_kwargs["on_chunk_done"] = _progress try: @@ -4034,26 +4071,30 @@ def _progress(idx: int, total: int, _result: dict) -> None: pg_result: dict = {"nodes": [], "edges": []} if cli_postgres_dsn is not None: from graphify.pg_introspect import introspect_postgres - print(f"[graphify extract] introspecting PostgreSQL schema...") + _extract_progress("introspecting PostgreSQL schema...") try: pg_result = introspect_postgres(cli_postgres_dsn) except (ConnectionError, ImportError) as exc: print(f"error: {exc}", file=sys.stderr) sys.exit(1) - print(f"[graphify extract] PostgreSQL: {len(pg_result['nodes'])} nodes, " - f"{len(pg_result['edges'])} edges") + _extract_progress( + f"PostgreSQL: {len(pg_result['nodes'])} nodes, " + f"{len(pg_result['edges'])} edges" + ) cargo_result: dict = {"nodes": [], "edges": []} if cli_cargo: from graphify.cargo_introspect import introspect_cargo - print("[graphify extract] introspecting Cargo workspace...") + _extract_progress("introspecting Cargo workspace...") try: cargo_result = introspect_cargo(target) except (ConnectionError, ImportError, OSError) as exc: print(f"error: {exc}", file=sys.stderr) sys.exit(1) - print(f"[graphify extract] Cargo: {len(cargo_result['nodes'])} nodes, " - f"{len(cargo_result['edges'])} edges") + _extract_progress( + f"Cargo: {len(cargo_result['nodes'])} nodes, " + f"{len(cargo_result['edges'])} edges" + ) # Merge AST + semantic + pg_result + cargo_result. Order matters for deduplication: passing AST # first means semantic node attributes win on collision (richer labels @@ -4318,7 +4359,8 @@ def _invalidate_file_manifest_for_db_graph() -> None: from graphify.export import to_json as _to_json from graphify.analyze import god_nodes as _god_nodes, surprising_connections as _surprising dedup_backend = backend if dedup_llm else None - if merge_existing_graph: + _extract_progress("building graph...") + if incremental_mode: # Prune everything the current scan no longer covers: genuinely # deleted manifest rows, excluded-but-alive manifest rows (#1908), # and the graph's own stale sources — which catches files that @@ -4372,8 +4414,12 @@ def _invalidate_file_manifest_for_db_graph() -> None: ) sys.exit(1) + _extract_progress( + f"clustering {G.number_of_nodes()} nodes, {G.number_of_edges()} edges..." + ) communities = _cluster(G, resolution=cli_resolution, exclude_hubs_percentile=cli_exclude_hubs) stages.mark("cluster") + _extract_progress("analyzing graph...") cohesion = _score_all(G, communities) try: gods = _god_nodes(G) @@ -4385,6 +4431,7 @@ def _invalidate_file_manifest_for_db_graph() -> None: surprises = [] stages.mark("analyze") + _extract_progress("writing graph.json...") from graphify.export import backup_if_protected as _backup _backup(graphify_out) _invalidate_file_manifest_for_db_graph() diff --git a/graphify/detect.py b/graphify/detect.py index 1adad00bb..138bb3f6c 100644 --- a/graphify/detect.py +++ b/graphify/detect.py @@ -7,6 +7,8 @@ import shlex import stat import subprocess +import sys +import threading import time import unicodedata from concurrent.futures import ThreadPoolExecutor @@ -19,7 +21,9 @@ convert_google_workspace_file, google_workspace_enabled, ) +from graphify.parallel import map_in_thread_pool, resolve_max_workers from graphify.paths import GRAPHIFY_OUT, out_path +from graphify.progress import verbose_enabled, vprint class FileType(str, Enum): @@ -47,6 +51,14 @@ class FileType(str, Enum): IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'} OFFICE_EXTENSIONS = {'.docx', '.xlsx'} VIDEO_EXTENSIONS = {'.mp4', '.mov', '.webm', '.mkv', '.avi', '.m4v', '.mp3', '.wav', '.m4a', '.ogg'} +_CORPUS_EXT_LOWER = frozenset( + ext.lower() + for group in ( + CODE_EXTENSIONS, DOC_EXTENSIONS, PAPER_EXTENSIONS, IMAGE_EXTENSIONS, + OFFICE_EXTENSIONS, VIDEO_EXTENSIONS, GOOGLE_WORKSPACE_EXTENSIONS, + ) + for ext in group +) CORPUS_WARN_THRESHOLD = 50_000 # words - below this, warn "you may not need a graph" CORPUS_UPPER_THRESHOLD = 500_000 # words - above this, warn about token cost @@ -807,6 +819,66 @@ def count_words(path: Path) -> int: return 0 +def _sum_word_counts(paths: list[Path], count_one: Callable[[Path], int]) -> int: + """Word-count every path; thread-pool when the corpus is large enough.""" + if not paths: + return 0 + mapped = map_in_thread_pool(count_one, paths) + if mapped is None: + return sum(count_one(p) for p in paths) + return sum(mapped) + + +class _Heartbeat: + """Rate-limited stderr progress so a long detect() walk is not silent (--verbose).""" + + def __init__(self, label: str, interval: float = 2.0) -> None: + self.label = label + self.interval = interval + self._t0 = time.monotonic() + self._last = self._t0 + self._n = 0 + self._lock = threading.Lock() + + def tick(self, n: int | None = None) -> None: + if not verbose_enabled(): + return + with self._lock: + self._n = n if n is not None else self._n + 1 + now = time.monotonic() + if (now - self._last) < self.interval and self._n % 2000 != 0: + return + self._last = now + elapsed = now - self._t0 + print( + f"[graphify] {self.label}: {self._n:,} ({elapsed:.0f}s)", + file=sys.stderr, + flush=True, + ) + + +class _WalkBucket: + __slots__ = ( + "files", "walk_errors", "ignored", "pruned_noise", + "skipped_sensitive", "nested_ignore", "nested_explicit", "top_dirs", + ) + + def __init__(self) -> None: + self.files: list[Path] = [] + self.walk_errors: list[str] = [] + self.ignored: list[str] = [] + self.pruned_noise: list[str] = [] + self.skipped_sensitive: list[str] = [] + self.nested_ignore: list[tuple[Path, str]] = [] + self.nested_explicit: list[tuple[Path, str]] = [] + self.top_dirs: list[str] = [] + + +def _keep_regular_file(path: Path) -> Path | None: + """Thread-pool entry: drop non-regular git-index paths.""" + return path if _is_regular_file(path) else None + + def _is_regular_file(path: Path) -> bool: """True only for regular files (symlinks followed). @@ -827,6 +899,7 @@ def _is_regular_file(path: Path) -> bool: _SKIP_DIRS = { "venv", ".venv", # "env"/".env"/"*_env" are gated on venv markers below (#2058) "node_modules", "__pycache__", ".git", + ".repo", # Google repo-tool metadata + object store; working trees sit beside it "dist", "build", "target", "out", "site-packages", "lib64", ".pytest_cache", ".mypy_cache", ".ruff_cache", @@ -922,10 +995,34 @@ def _has_venv_markers(d: "Path") -> bool: return False +def _is_buildroot_output(d: "Path") -> bool: + """True when *d* is a Buildroot ``output/`` tree (host sysroot, not source).""" + try: + names = set(os.listdir(d)) + except OSError: + return False + return {"host", "target"} <= names or {"host", "images"} <= names or {"images", "staging"} <= names + + +def _maybe_code_filename(name: str) -> bool: + """Keep classifiable (and extensionless/shebang) names for --code-only scans.""" + dot = name.rfind(".") + if dot <= 0: + return True + return name[dot:].lower() in _CORPUS_EXT_LOWER + + def _is_noise_dir(part: str, parent: "Path | None" = None) -> bool: """Return True if this directory name looks like a venv, cache, or dep dir.""" if part in _SKIP_DIRS: return True + if part == "output" or part.startswith("output_"): + # Buildroot writes millions of files under output/ / output_/. + # A source dir that happens to be named output stays unless it has + # host/+target/ (or images/) layout. + if parent is None: + return False + return _is_buildroot_output(parent / part) if part in ("env", ".env") or part.endswith("_env"): # Ambiguous: a real venv OR a real source dir. Prune only on actual venv # evidence, mirroring the "snapshots" gating (#1666/#2058). @@ -1025,6 +1122,434 @@ def _path_identity(path: Path) -> str: return _nfc(os.path.normcase(os.path.abspath(os.fspath(path)))) +# `git ls-files` of a million-path index is still seconds; 30s was too tight +# and forced a Python os.walk fallback on the repos that need Git most. +_GIT_LS_TIMEOUT = 300 + + +def _git_ls_z( + vcs_root: Path, + extra_args: list[str], + pathspec: str | None, +) -> list[str] | None: + """NUL-delimited ``git ls-files`` paths relative to *vcs_root*, or None.""" + cmd = ["git", "-C", str(vcs_root), "ls-files", "-z", *extra_args] + if pathspec: + cmd.extend(["--", pathspec]) + try: + proc = subprocess.run( + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + check=False, + timeout=_GIT_LS_TIMEOUT, + env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, + ) + except (OSError, subprocess.SubprocessError): + return None + if proc.returncode != 0: + return None + return [os.fsdecode(raw) for raw in proc.stdout.split(b"\0") if raw] + + +def _find_repo_manifest_root(start: Path) -> Path | None: + """Nearest ancestor (inclusive) with a Google ``repo`` ``.repo/project.list``.""" + current = start.resolve() + home = Path.home() + while True: + if (current / ".repo" / "project.list").is_file(): + return current + parent = current.parent + if parent == current or current == home: + return None + current = parent + + +def _repo_project_rels(repo_root: Path) -> list[str] | None: + listing = repo_root / ".repo" / "project.list" + try: + lines = listing.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return None + projects = [ln.strip() for ln in lines if ln.strip() and not ln.startswith("#")] + return projects or None + + +def _ingest_git_relpaths( + *, + root: Path, + join_root: Path, + cached: list[str], + others: list[str], + code_only: bool, + files: list[Path], + seen: set[str], + tracked_files: set[str], + pruned: list[str], + pruned_seen: set[str], +) -> None: + """Append Git-listed paths under *join_root* onto the shared scan buckets.""" + root_s = os.path.abspath(os.fspath(root)) + root_prefix = root_s + os.sep + join_s = os.path.abspath(os.fspath(join_root)) + + def _take(rel_s: str, tracked: bool) -> None: + path_s = os.path.abspath(os.path.join(join_s, rel_s.replace("/", os.sep))) + if path_s != root_s and not path_s.startswith(root_prefix): + return + name = os.path.basename(path_s) + if name in _SKIP_FILES: + return + if code_only and not _maybe_code_filename(name): + return + if os.path.isdir(path_s): + # Superproject indexes list submodule gitlinks as paths; the + # checkout is a directory and is enumerated from that worktree. + return + nest = os.path.dirname(path_s) + join_prefix = join_s + os.sep + while nest.startswith(join_prefix): + if os.path.exists(os.path.join(nest, ".git")): + return + nest = os.path.dirname(nest) + rel_parts = rel_s.replace("\\", "/").split("/") + parent = join_root + for part in rel_parts[:-1]: + if _is_noise_dir(part, parent): + key = str(parent / part) + os.sep + if key not in pruned_seen: + pruned_seen.add(key) + pruned.append(key) + return + parent = parent / part + ident = _nfc(os.path.normcase(path_s)) + if ident in seen: + return + seen.add(ident) + files.append(Path(path_s)) + if tracked: + tracked_files.add(ident) + + for rel_s in cached: + _take(rel_s, True) + for rel_s in others: + _take(rel_s, False) + + +def _git_enumerate_single_repo( + root: Path, + vcs_root: Path, + *, + code_only: bool, +) -> tuple[list[Path], set[str], set[str], list[str]] | None: + try: + rel = root.resolve().relative_to(vcs_root.resolve()).as_posix() + except ValueError: + return None + pathspec = None if rel in (".", "") else rel + vprint( + f"listing files with git ls-files " + f"({pathspec or '.'}) ..." + ) + cached = _git_ls_z(vcs_root, ["--cached"], pathspec) + if cached is None: + return None + others = _git_ls_z(vcs_root, ["--others", "--exclude-standard"], pathspec) + if others is None: + return None + files: list[Path] = [] + seen: set[str] = set() + tracked_files: set[str] = set() + pruned: list[str] = [] + pruned_seen: set[str] = set() + _ingest_git_relpaths( + root=root, join_root=vcs_root, cached=cached, others=others, + code_only=code_only, files=files, seen=seen, + tracked_files=tracked_files, pruned=pruned, pruned_seen=pruned_seen, + ) + return files, tracked_files, set(), pruned + + +def _ls_git_worktree(join_root_s: str) -> tuple[str, list[str], list[str]] | None: + """Thread-pool worker: ``git ls-files`` one working tree.""" + proj = Path(join_root_s) + if not (proj / ".git").exists(): + return None + cached = _git_ls_z(proj, ["--cached"], None) + others = _git_ls_z(proj, ["--others", "--exclude-standard"], None) + if cached is None or others is None: + return None + return join_root_s, cached, others + + +def _parse_gitmodules_paths(gitmodules: Path) -> list[str]: + """``path =`` entries from a ``.gitmodules`` file.""" + try: + text = gitmodules.read_text(encoding="utf-8", errors="replace") + except OSError: + return [] + paths: list[str] = [] + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith("#") or line.startswith(";"): + continue + if line.lower().startswith("path") and "=" in line: + rel = line.split("=", 1)[1].strip().replace("\\", "/") + if rel: + paths.append(rel) + return paths + + +def _gitmodules_paths(super_root: Path) -> list[str]: + """Submodule paths relative to *super_root*, including nested checkouts.""" + gitmodules = super_root / ".gitmodules" + if not gitmodules.is_file(): + return [] + paths: list[str] = [] + try: + proc = subprocess.run( + ["git", "-C", str(super_root), "submodule", "status", "--recursive"], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + check=False, + timeout=_GIT_LS_TIMEOUT, + env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, + ) + except (OSError, subprocess.SubprocessError): + proc = None + if proc is not None and proc.returncode == 0 and proc.stdout: + for raw in proc.stdout.splitlines(): + line = raw.strip() + if not line: + continue + parts = line.split() + if len(parts) >= 2: + paths.append(parts[1].replace("\\", "/")) + if not paths: + paths = _parse_gitmodules_paths(gitmodules) + nested: list[str] = [] + for rel in paths: + nested.extend( + f"{rel}/{p}" + for p in _parse_gitmodules_paths(super_root / rel / ".gitmodules") + ) + paths.extend(nested) + seen: set[str] = set() + ordered: list[str] = [] + for rel in paths: + rel = rel.strip("/") + if not rel or rel in seen: + continue + seen.add(rel) + ordered.append(rel) + return ordered + + +def _submodule_super_for_scan(root: Path) -> Path | None: + """Superproject to enumerate when this scan should use git submodules. + + Only the scan root's own ``.gitmodules``, or an ancestor superproject when + the scan root is *not* already a working tree (so scanning one submodule + does not pull in its siblings). + """ + root = root.resolve() + if (root / ".gitmodules").is_file() and _gitmodules_paths(root): + return root + if (root / ".git").exists(): + return None + vcs = _find_vcs_root(root) + if vcs is not None and (vcs / ".gitmodules").is_file() and _gitmodules_paths(vcs): + return vcs + return None + + +def _path_is_under(inner: Path, outer: Path) -> bool: + try: + inner.resolve().relative_to(outer.resolve()) + return True + except (ValueError, OSError): + return False + + +def _worktree_overlaps_scan(worktree: Path, root: Path) -> bool: + return _path_is_under(worktree, root) or _path_is_under(root, worktree) + + +def _git_enumerate_named_worktrees( + root: Path, + worktrees: list[Path], + *, + code_only: bool, + kind: str, +) -> tuple[list[Path], set[str], set[str], list[str]] | None: + """``git ls-files`` each working tree and merge under the scan root.""" + uniq: list[Path] = [] + seen: set[str] = set() + for w in worktrees: + try: + resolved = w.resolve() + except OSError: + resolved = w + if not _worktree_overlaps_scan(resolved, root): + continue + if not (resolved / ".git").exists(): + continue + ident = _path_identity(resolved) + if ident in seen: + continue + seen.add(ident) + uniq.append(resolved) + if not uniq: + return None + vprint(f"{kind}: git ls-files in {len(uniq)} worktree(s) ...") + payloads = [str(p) for p in uniq] + listed = map_in_thread_pool(_ls_git_worktree, payloads, threshold=2) + if listed is None: + listed = [_ls_git_worktree(p) for p in payloads] + files: list[Path] = [] + seen_files: set[str] = set() + tracked_files: set[str] = set() + pruned: list[str] = [] + pruned_seen: set[str] = set() + ok = 0 + for row in listed: + if row is None: + continue + ok += 1 + join_s, cached, others = row + _ingest_git_relpaths( + root=root, join_root=Path(join_s), + cached=cached, others=others, code_only=code_only, + files=files, seen=seen_files, tracked_files=tracked_files, + pruned=pruned, pruned_seen=pruned_seen, + ) + if ok == 0: + return None + vprint( + f"{kind}: listed {len(files):,} files " + f"from {ok}/{len(uniq)} worktree(s)" + ) + return files, tracked_files, set(), pruned + + +def _find_git_worktrees( + root: Path, + *, + dir_ignored: Callable[[Path], bool] | None = None, +) -> list[Path]: + """Working trees under *root* (directories that contain a ``.git`` marker). + + Descends with noise pruning (``.repo``, Buildroot ``output_*``, …) and the + same ignore predicate as the corpus walk, and does not follow directory + symlinks. Nested checkouts are included. + """ + vprint("searching for .git folders ...") + found: list[Path] = [] + for dirpath, dirnames, _filenames in os.walk(root, followlinks=False): + dp = Path(dirpath) + kept: list[str] = [] + for d in dirnames: + if d == ".git": + continue + child = dp / d + if _is_noise_dir(d, dp): + continue + if dir_ignored is not None: + try: + if dir_ignored(child): + continue + except (OSError, ValueError): + continue + if os.path.islink(os.fspath(child)): + continue + kept.append(d) + dirnames[:] = kept + if (dp / ".git").exists(): + found.append(dp) + found.sort(key=lambda p: (len(p.parts), str(p))) + return found + + +def _git_enumerate_repo_projects( + root: Path, + repo_root: Path, + *, + code_only: bool, +) -> tuple[list[Path], set[str], set[str], list[str]] | None: + projects = _repo_project_rels(repo_root) + if not projects: + return None + try: + scan_rel = root.resolve().relative_to(repo_root.resolve()).as_posix() + except ValueError: + return None + if scan_rel not in (".", ""): + prefix = scan_rel + "/" + projects = [ + p for p in projects + if p == scan_rel or p.startswith(prefix) + ] + if not projects: + return None + return _git_enumerate_named_worktrees( + root, + [repo_root / p for p in projects], + code_only=code_only, + kind="repo workspace", + ) + + +def _git_enumerate_files( + root: Path, + *, + code_only: bool = False, + dir_ignored: Callable[[Path], bool] | None = None, +) -> tuple[list[Path], set[str], set[str], list[str]] | None: + """List discoverable files via Git instead of ``os.walk``. + + Order: + 1. Git submodules (``.gitmodules`` at the scan root, or an ancestor + superproject when the scan root is not itself a working tree). + 2. Google ``repo`` workspace (``.repo/project.list``). + 3. Every nested ``.git`` working tree under the scan root. + 4. A single ancestor Git repo with a pathspec (subfolder of a normal repo). + """ + super_root = _submodule_super_for_scan(root) + if super_root is not None: + sub_rels = _gitmodules_paths(super_root) + if sub_rels: + worktrees = [super_root, *[super_root / rel for rel in sub_rels]] + listed = _git_enumerate_named_worktrees( + root, worktrees, code_only=code_only, kind="git submodules", + ) + if listed is not None: + return listed + repo_root = _find_repo_manifest_root(root) + if repo_root is not None: + listed = _git_enumerate_repo_projects(root, repo_root, code_only=code_only) + if listed is not None: + return listed + if (root / ".git").exists(): + listed = _git_enumerate_named_worktrees( + root, [root], code_only=code_only, kind="git", + ) + if listed is not None: + return listed + worktrees = _find_git_worktrees(root, dir_ignored=dir_ignored) + if worktrees: + listed = _git_enumerate_named_worktrees( + root, worktrees, code_only=code_only, kind="nested git", + ) + if listed is not None: + return listed + vcs_root = _find_vcs_root(root) + if vcs_root is not None and (vcs_root / ".git").exists(): + return _git_enumerate_single_repo(root, vcs_root, code_only=code_only) + return None + + def _git_tracked_path_keys(root: Path) -> tuple[set[str], set[str]]: """Return tracked-file keys and their ancestor-directory keys under *root*. @@ -1045,7 +1570,7 @@ def _git_tracked_path_keys(root: Path) -> tuple[set[str], set[str]]: stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False, - timeout=30, + timeout=_GIT_LS_TIMEOUT, env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, ) except (OSError, subprocess.SubprocessError): @@ -1063,11 +1588,8 @@ def _git_tracked_path_keys(root: Path) -> tuple[set[str], set[str]]: path.relative_to(root) except ValueError: continue - # Deleted index entries and submodule gitlinks are not discoverable - # files. Symlinks to regular files remain eligible; the existing - # in-root target guard still decides whether they may enter the corpus. - if not _is_regular_file(path): - continue + # Names from the index, no extra stat: regularity is checked at admit + # time. Statting every cached path doubled scan cost on huge repos. tracked_files.add(_path_identity(path)) parent = path.parent while parent != root: @@ -1662,7 +2184,7 @@ def _resolves_under_root(path: Path, root: Path) -> bool: return True -def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace: bool | None = None, extra_excludes: list[str] | None = None, cache_root: Path | None = None, gitignore: bool = True) -> dict: +def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace: bool | None = None, extra_excludes: list[str] | None = None, cache_root: Path | None = None, gitignore: bool = True, code_only: bool = False) -> dict: root = root.resolve() configured_out_dir = root / GRAPHIFY_OUT configured_out_names = {configured_out_dir.name} @@ -1694,7 +2216,7 @@ def detect(root: Path, *, follow_symlinks: bool | None = None, google_workspace: FileType.IMAGE: [], FileType.VIDEO: [], } - total_words = 0 + pending_wc: list[Path] = [] def _wc(path: Path) -> int: # Cache word counts against each file's stat signature so unchanged @@ -1714,15 +2236,16 @@ def _wc(path: Path) -> int: pruned_noise: list[str] = [] ignore_patterns = _load_graphifyignore(root, gitignore=gitignore) explicit_ignore_patterns = _load_graphifyignore(root, gitignore=False) - # See ignored_predicate: skip the `git ls-files` subprocess when .gitignore - # contributes no patterns, so a non-.gitignore corpus pays nothing for it. - tracked_files, tracked_dirs = ( - _git_tracked_path_keys(root) - if gitignore and len(ignore_patterns) > len(explicit_ignore_patterns) - else (set(), set()) - ) + tracked_files: set[str] = set() + tracked_dirs: set[str] = set() ignore_cache: dict[Path, bool] = {} # shared across all _is_ignored calls in this scan explicit_ignore_cache: dict[Path, bool] = {} + # Positive directory-ignore memo: a nested-git hunt and the later os.walk + # both ask about the same ignored dir. Counting each as a fresh + # ``_is_scan_ignored`` call is the 29k-file defect shape the tests guard. + # Only True is sticky — False must be rechecked after a nested ignore file + # is loaded (#1922). + ignored_dirs: set[Path] = set() # CLI --exclude patterns are anchored at the scan root and appended last # so they win over any .graphifyignore/.gitignore rules (#947). if extra_excludes: @@ -1733,7 +2256,9 @@ def _wc(path: Path) -> int: explicit_ignore_patterns.append((root, line)) def _ignored_for_scan(path: Path) -> bool: - return _is_scan_ignored( + if path in ignored_dirs: + return True + val = _is_scan_ignored( path, root, ignore_patterns, @@ -1743,47 +2268,77 @@ def _ignored_for_scan(path: Path) -> bool: cache=ignore_cache, explicit_cache=explicit_ignore_cache, ) + if val: + ignored_dirs.add(path) + return val # Always include graphify-out/memory/ - query results filed back into the graph memory_dir = root / GRAPHIFY_OUT / "memory" - scan_paths = [root] - if memory_dir.exists(): - scan_paths.append(memory_dir) - seen: set[Path] = set() - all_files: list[Path] = [] - - # os.walk swallows os.scandir errors by default (no onerror -> the failing - # directory subtree is silently skipped). That turns a transient - # PermissionError, or a directory created/deleted mid-walk (e.g. concurrent - # writes racing the scan), into a partial file list and, downstream, a - # silently partial graph.json. Record and surface every skipped directory - # so an incomplete enumeration is visible rather than silent. - walk_errors: list[str] = [] + vprint(f"scanning {root} ...") + heartbeat = _Heartbeat("scanning") + + def _walk_from( + start: Path, + *, + in_memory: bool, + patterns: list[tuple[Path, str]], + explicit: list[tuple[Path, str]], + descend: bool, + ) -> _WalkBucket: + """Walk *start*. Nested ignore files are loaded onto local copies. + + Sibling subtrees can run this in parallel: each call gets a snapshot of + ancestor patterns, and a pattern only matches under its own anchor. + """ + patterns = list(patterns) + explicit = list(explicit) + base_ignore = len(patterns) + base_explicit = len(explicit) + cache: dict[Path, bool] = {} + explicit_cache: dict[Path, bool] = {} + bucket = _WalkBucket() + local_seen: set[Path] = set() + + def _ignored_local(path: Path) -> bool: + # Nested .gitignore/.graphifyignore is loaded onto *patterns* as + # the walk enters each directory. The outer _ignored_for_scan + # closure still sees only ancestor rules, so directory pruning + # MUST use this local copy or a sibling project's ignore file + # is applied too late (file-level drop instead of a recorded + # ignored subtree, #1922). + if path in ignored_dirs: + return True + val = _is_scan_ignored( + path, root, patterns, explicit, + tracked_files, tracked_dirs, + cache=cache, explicit_cache=explicit_cache, + ) + if val: + ignored_dirs.add(path) + return val - def _on_walk_error(err: OSError) -> None: - import sys as _sys - target = getattr(err, "filename", None) or "" - walk_errors.append(f"{target}: {err}") - print( - f"[graphify] WARNING: could not scan {target} ({err}); " - f"its files are missing from this run's enumeration.", - file=_sys.stderr, - ) + def _on_err(err: OSError) -> None: + target = getattr(err, "filename", None) or "" + bucket.walk_errors.append(f"{target}: {err}") + print( + f"[graphify] WARNING: could not scan {target} ({err}); " + f"its files are missing from this run's enumeration.", + file=sys.stderr, flush=True, + ) - for scan_root in scan_paths: - in_memory_tree = memory_dir.exists() and str(scan_root).startswith(str(memory_dir)) for dirpath, dirnames, filenames in os.walk( - scan_root, followlinks=follow_symlinks, onerror=_on_walk_error + start, followlinks=follow_symlinks, onerror=_on_err, ): dp = Path(dirpath) + heartbeat.tick() if follow_symlinks and os.path.islink(dirpath): real = os.path.realpath(dirpath) parent_real = os.path.realpath(os.path.dirname(dirpath)) if parent_real == real or parent_real.startswith(real + os.sep): dirnames.clear() continue - if not in_memory_tree: + if not in_memory: # dp == root was already loaded by _load_graphifyignore (root is # the last entry in its ancestor chain); every other directory # reached by the walk is a descendant below the scan root, whose @@ -1791,10 +2346,8 @@ def _on_walk_error(err: OSError) -> None: # Load it now, before pruning dp's children, so a nested ignore # file governs its own subtree the same way git honors it (#1206). if dp != root: - ignore_patterns.extend(_load_dir_own_ignore(dp, gitignore=gitignore)) - explicit_ignore_patterns.extend( - _load_dir_own_ignore(dp, gitignore=False) - ) + patterns.extend(_load_dir_own_ignore(dp, gitignore=gitignore)) + explicit.extend(_load_dir_own_ignore(dp, gitignore=False)) # Prune noise dirs in-place so os.walk never descends into them. # Dot dirs are allowed — users often want .github/, .claude/, etc. # Framework caches (.next, .nuxt, …) are caught by _is_noise_dir. @@ -1817,57 +2370,178 @@ def _on_walk_error(err: OSError) -> None: except (OSError, RuntimeError): pass if is_configured_out: - pruned_noise.append(str(child) + os.sep) + bucket.pruned_noise.append(str(child) + os.sep) continue if _is_noise_dir(d, dp): # Record pruned-as-noise dirs so a wrongly-pruned real # source dir is at least traceable in the output rather # than vanishing silently (#2058). - pruned_noise.append(str(child) + os.sep) + bucket.pruned_noise.append(str(child) + os.sep) continue # Directory-level pruning: ONE ignore evaluation excludes the # whole subtree — os.walk never descends, so a 29k-file # ignored dir costs one check, not one per contained file. - if _ignored_for_scan(child): - ignored.append(str(child) + os.sep) + if _ignored_local(child): + bucket.ignored.append(str(child) + os.sep) continue kept_dirs.append(d) dirnames[:] = kept_dirs - if follow_symlinks: - safe_dirs: list[str] = [] - for d in dirnames: - child = dp / d - if child.is_symlink() and not _resolves_under_root(child, root): - skipped_sensitive.append(str(child) + " [symlink target outside scan root]") - continue - safe_dirs.append(d) - dirnames[:] = safe_dirs + # Out-of-root symlink dirs must be pruned even in the memory walk + # (which skips ignore/noise). Otherwise followlinks=True descends + # into the target and admits its regular files, which are not + # themselves symlinks so a file-islink check cannot catch them. + if follow_symlinks: + safe_dirs: list[str] = [] + for d in dirnames: + child = dp / d + if child.is_symlink() and not _resolves_under_root(child, root): + bucket.skipped_sensitive.append( + str(child) + " [symlink target outside scan root]" + ) + continue + safe_dirs.append(d) + dirnames[:] = safe_dirs + if not in_memory: + if not descend and dp == start: + if follow_symlinks: + bucket.top_dirs = list(dirnames) + else: + # os.walk(followlinks=False) lists symlink dirs but + # does not recurse into them. Walking the symlink as + # a subtree start WOULD follow it. + bucket.top_dirs = [ + d for d in dirnames if not os.path.islink(str(dp / d)) + ] + dirnames.clear() for fname in filenames: if fname in _SKIP_FILES: continue + if code_only and not _maybe_code_filename(fname): + continue p = dp / fname - if p not in seen: - seen.add(p) - all_files.append(p) + if p not in local_seen: + local_seen.add(p) + bucket.files.append(p) + bucket.nested_ignore = patterns[base_ignore:] + bucket.nested_explicit = explicit[base_explicit:] + return bucket + + def _merge_walk(bucket: _WalkBucket) -> None: + ignore_patterns.extend(bucket.nested_ignore) + explicit_ignore_patterns.extend(bucket.nested_explicit) + walk_errors.extend(bucket.walk_errors) + ignored.extend(bucket.ignored) + pruned_noise.extend(bucket.pruned_noise) + skipped_sensitive.extend(bucket.skipped_sensitive) + for p in bucket.files: + if p not in seen: + seen.add(p) + all_files.append(p) + + seen: set[Path] = set() + all_files: list[Path] = [] + walk_errors: list[str] = [] + + # Walk the scan root's own directory first (no descend) so sibling + # subtrees can be scanned concurrently. Each subtree copies ancestor + # ignore patterns; a nested .gitignore only matches under its anchor. + # Prefer Git's C walker when we can: it never enters gitignored trees + # (the usual million-file ``node_modules/`` case). Python os.walk cannot. + _no_git_enum = os.environ.get("GRAPHIFY_NO_GIT_ENUM", "").strip().lower() in ( + "1", "true", "yes", + ) + git_enum = None + if gitignore and not follow_symlinks and not _no_git_enum: + git_enum = _git_enumerate_files( + root, code_only=code_only, dir_ignored=_ignored_for_scan, + ) + + if git_enum is not None: + all_files, tracked_files, tracked_dirs, git_pruned = git_enum + pruned_noise.extend(git_pruned) + out_prefix = str(configured_out_dir) + os.sep + mem_prefix = str(memory_dir) + os.sep + all_files = [ + p for p in all_files + if not str(p).startswith(out_prefix) or str(p).startswith(mem_prefix) + ] + vprint( + f"git listed {len(all_files):,} files " + f"(ignored trees not walked)" + ) + # Git already applied gitignore. Only load extra .graphifyignore files + # along listed paths — do not re-read every nested .gitignore. + seen_gi: set[str] = set() + root_s = str(root) + root_prefix = root_s + os.sep + for p in all_files: + parent = os.path.dirname(str(p)) + while parent.startswith(root_prefix) or parent == root_s: + if parent == root_s: + break + if parent in seen_gi: + break + seen_gi.add(parent) + if os.path.isfile(os.path.join(parent, ".graphifyignore")): + extra = _load_dir_own_ignore(Path(parent), gitignore=False) + ignore_patterns.extend(extra) + explicit_ignore_patterns.extend(extra) + parent = os.path.dirname(parent) + else: + vprint("walking filesystem (no git/repo index) ...") + if gitignore and len(ignore_patterns) > len(explicit_ignore_patterns): + tracked_files, tracked_dirs = _git_tracked_path_keys(root) + root_bucket = _walk_from( + root, in_memory=False, + patterns=ignore_patterns, explicit=explicit_ignore_patterns, + descend=False, + ) + _merge_walk(root_bucket) + top_paths = [root / d for d in root_bucket.top_dirs] + snapshot_ignore = list(ignore_patterns) + snapshot_explicit = list(explicit_ignore_patterns) + + def _walk_child(start: Path) -> _WalkBucket: + return _walk_from( + start, in_memory=False, + patterns=snapshot_ignore, explicit=snapshot_explicit, + descend=True, + ) + + child_parts = None + if len(top_paths) >= 2 and resolve_max_workers(len(top_paths), None) > 1: + child_parts = map_in_thread_pool(_walk_child, top_paths, threshold=2) + if child_parts is None: + child_parts = [_walk_child(p) for p in top_paths] + for part in child_parts: + _merge_walk(part) + + if memory_dir.exists(): + _merge_walk(_walk_from( + memory_dir, in_memory=True, + patterns=ignore_patterns, explicit=explicit_ignore_patterns, + descend=True, + )) all_files.sort(key=lambda p: str(p)) out_base = Path(cache_root).resolve() if cache_root is not None else root converted_dir = out_base / GRAPHIFY_OUT / "converted" - for p in all_files: - # For memory dir files, skip hidden/noise filtering + def _admit(p: Path) -> tuple: + """Classify one walked path. Return a tagged result tuple.""" + heartbeat.tick() in_memory = memory_dir.exists() and str(p).startswith(str(memory_dir)) if not in_memory: - # Skip files inside our own converted/ dir (avoid re-processing sidecars) if str(p).startswith(str(converted_dir)): - continue + return ("skip",) if not in_memory and _ignored_for_scan(p): - ignored.append(str(p)) - continue + return ("ignored", str(p)) + # Any path whose resolve() sits outside root — symlink file, or a + # regular file reached by following a symlink directory. v8 checked + # this for every candidate; gating on islink() admitted the latter. if not _resolves_under_root(p, root): - skipped_sensitive.append(str(p) + " [symlink target outside scan root]") - continue + return ("sensitive", str(p) + " [symlink target outside scan root]") if not _is_regular_file(p): # A repository may contain named pipes, sockets and device nodes, # and `clone ` exists precisely to point the scan at @@ -1878,76 +2552,105 @@ def _on_walk_error(err: OSError) -> None: # number in the hundreds across the extractors, and open() on a # FIFO with no writer BLOCKS FOREVER — it never raises, so their # try/except cannot help and the whole run hangs with no output. - skipped_sensitive.append(str(p) + " [not a regular file]") - continue + return ("sensitive", str(p) + " [not a regular file]") if _is_sensitive(p): - skipped_sensitive.append(str(p)) - continue + return ("sensitive", str(p)) ftype = classify_file(p) if not ftype: # Considered but unclassifiable: an extension not in any supported set, # or an extensionless, non-shebang file (Dockerfile, Gemfile, Makefile, # Rakefile, LICENSE, ...). Previously these left no trace at all — not # counted, not listed — so a user couldn't tell they were seen (#1692). - unclassified.append(str(p)) + return ("unclassified", str(p)) + if p.suffix.lower() in GOOGLE_WORKSPACE_EXTENSIONS: + if code_only: + return ("file", ftype, str(p), None) + if not google_workspace: + return ( + "sensitive", + str(p) + + " [Google Workspace shortcut skipped - pass --google-workspace " + "or set GRAPHIFY_GOOGLE_WORKSPACE=1]", + ) + try: + md_path = convert_google_workspace_file( + p, converted_dir, xlsx_to_markdown=xlsx_to_markdown, root=root, + ) + except Exception as exc: + return ("sensitive", str(p) + f" [Google Workspace export failed: {exc}]") + if md_path: + if _ignored_for_scan(md_path): + return ("skip",) + return ("file", ftype, str(md_path), md_path) + return ("sensitive", str(p) + " [Google Workspace export produced no readable text]") + if p.suffix.lower() in OFFICE_EXTENSIONS: + if code_only: + return ("file", ftype, str(p), None) + md_path = convert_office_file(p, converted_dir, root=root) + if md_path: + if _ignored_for_scan(md_path): + return ("skip",) + return ("file", ftype, str(md_path), md_path) + return ( + "sensitive", + str(p) + " [office conversion failed - pip install graphifyy[office]]", + ) + wc = None if code_only or ftype == FileType.VIDEO else p + return ("file", ftype, str(p), wc) + + admitted = map_in_thread_pool(_admit, all_files) + if admitted is None: + admitted = [_admit(p) for p in all_files] + for row in admitted: + kind = row[0] + if kind == "skip": continue - if ftype: - if p.suffix.lower() in GOOGLE_WORKSPACE_EXTENSIONS: - if not google_workspace: - skipped_sensitive.append( - str(p) - + " [Google Workspace shortcut skipped - pass --google-workspace " - "or set GRAPHIFY_GOOGLE_WORKSPACE=1]" - ) - continue - try: - md_path = convert_google_workspace_file(p, converted_dir, xlsx_to_markdown=xlsx_to_markdown, root=root) - except Exception as exc: - skipped_sensitive.append(str(p) + f" [Google Workspace export failed: {exc}]") - continue - if md_path: - if _ignored_for_scan(md_path): - continue - files[ftype].append(str(md_path)) - total_words += _wc(md_path) - else: - skipped_sensitive.append(str(p) + " [Google Workspace export produced no readable text]") - continue - # Office files: convert to markdown sidecar so subagents can read them - if p.suffix.lower() in OFFICE_EXTENSIONS: - md_path = convert_office_file(p, converted_dir, root=root) - if md_path: - if _ignored_for_scan(md_path): - continue - files[ftype].append(str(md_path)) - total_words += _wc(md_path) - else: - # Conversion failed (library not installed) - skip with note - skipped_sensitive.append(str(p) + " [office conversion failed - pip install graphifyy[office]]") - continue - files[ftype].append(str(p)) - if ftype != FileType.VIDEO: - total_words += _wc(p) + if kind == "ignored": + ignored.append(row[1]) + continue + if kind == "sensitive": + skipped_sensitive.append(row[1]) + continue + if kind == "unclassified": + unclassified.append(row[1]) + continue + _, ftype, stored, wc = row + files[ftype].append(stored) + if wc is not None: + pending_wc.append(wc) for ftype in files: files[ftype].sort() total_files = sum(len(v) for v in files.values()) - needs_graph = total_words >= CORPUS_WARN_THRESHOLD - - # Determine warning - lower bound, upper bound, or sensitive files skipped - warning: str | None = None - if not needs_graph: - warning = ( - f"Corpus is ~{total_words:,} words - fits in a single context window. " - f"You may not need a graph." - ) - elif total_words >= CORPUS_UPPER_THRESHOLD or total_files >= FILE_COUNT_UPPER: - warning = ( - f"Large corpus: {total_files} files · ~{total_words:,} words. " - f"Semantic extraction will be expensive (many Claude tokens). " - f"Consider running on a subfolder." - ) + if code_only: + # Opening every file just to size the corpus dominates on huge trees; + # --code-only only needs AST membership, not a word budget. + total_words = 0 + needs_graph = total_files > 0 + warning = None + if total_files >= FILE_COUNT_UPPER: + warning = ( + f"Large corpus: {total_files} files (word count skipped, --code-only). " + f"Semantic extraction is off; AST indexing still scales with file count." + ) + else: + if len(pending_wc) >= 20: + vprint(f"counting words in {len(pending_wc):,} files ...") + total_words = _sum_word_counts(pending_wc, _wc) + needs_graph = total_words >= CORPUS_WARN_THRESHOLD + warning = None + if not needs_graph: + warning = ( + f"Corpus is ~{total_words:,} words - fits in a single context window. " + f"You may not need a graph." + ) + elif total_words >= CORPUS_UPPER_THRESHOLD or total_files >= FILE_COUNT_UPPER: + warning = ( + f"Large corpus: {total_files} files · ~{total_words:,} words. " + f"Semantic extraction will be expensive (many Claude tokens). " + f"Consider running on a subfolder." + ) return { "files": {k.value: v for k, v in files.items()}, @@ -2372,6 +3075,7 @@ def detect_incremental( kind: str = "semantic", extra_excludes: list[str] | None = None, gitignore: bool = True, + code_only: bool = False, ) -> dict: """Like detect(), but returns only new or modified files since the last run. @@ -2401,6 +3105,7 @@ def detect_incremental( google_workspace=google_workspace, extra_excludes=extra_excludes, gitignore=gitignore, + code_only=code_only, ) # Pass ``root`` so a manifest written with relative keys (post-#777) is # re-anchored to the absolute form the rest of this function compares diff --git a/graphify/extract.py b/graphify/extract.py index e015c9d71..9e129e465 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -16,6 +16,14 @@ from .cache import load_cached, save_cached from .mcp_ingest import extract_mcp_config, is_mcp_config_path from .manifest_ingest import extract_package_manifest, is_package_manifest_path +from .parallel import ( + PARALLEL_THRESHOLD, + chunk_size_for, + chunked, + map_in_thread_pool, + resolve_max_workers, +) +from .progress import vprint from .resolver_registry import ( LanguageResolver, register as register_language_resolver, @@ -5700,6 +5708,57 @@ def _extract_single_file(args: tuple) -> tuple[int, dict]: return idx, result +def _extract_file_batch(items: list[tuple]) -> list[tuple[int, dict]]: + """Process-pool entry: extract a chunk of files in one submitted task. + + One Future per file made a 50k-file corpus spawn 50k pickles; batching + keeps worker count on the CPU and cuts scheduling overhead. + """ + return [_extract_single_file(item) for item in items] + + +def _probe_one_cached(args: tuple[int, str, str, str]) -> tuple[int, dict | None]: + """Cache probe for one file. ``None`` result means the file still needs extract. + + Thread-pool entry (I/O + JSON). Must stay pickle-free nested-friendly; it is + module-level so ``map_in_thread_pool`` can also be swapped for a process + pool later without changing the payload. + """ + i, path_str, root_str, cache_loc_str = args + path = Path(path_str) + if _get_extractor(path) is None: + return i, {"nodes": [], "edges": []} + if path.suffix in _JS_CACHE_BYPASS_SUFFIXES: + return i, None + return i, load_cached(path, Path(root_str), cache_root=Path(cache_loc_str)) + + +def _probe_cached_files( + paths: list[Path], + root: Path, + cache_location: Path, + *, + parallel: bool, + max_workers: int | None, +) -> tuple[list[dict | None], list[tuple[int, Path]]]: + """Fill per-file cache hits; return (per_file, uncached_work) in path order.""" + total = len(paths) + per_file: list[dict | None] = [None] * total + root_str = str(root) + cache_loc_str = str(cache_location) + items = [(i, str(path), root_str, cache_loc_str) for i, path in enumerate(paths)] + + mapped = None + if parallel: + mapped = map_in_thread_pool(_probe_one_cached, items, max_workers=max_workers) + if mapped is None: + mapped = [_probe_one_cached(item) for item in items] + for i, result in mapped: + per_file[i] = result + uncached_work = [(i, paths[i]) for i in range(total) if per_file[i] is None] + return per_file, uncached_work + + def _extract_parallel( uncached_work: list[tuple[int, Path]], per_file: list[dict | None], @@ -5717,31 +5776,7 @@ def _extract_parallel( """ import concurrent.futures - if max_workers is None: - # Honour GRAPHIFY_MAX_WORKERS env override; otherwise scale to the - # full CPU. The historical `, 8)` cap was a safety bound for laptops - # in 2023 — on a 32-thread workstation it costs a 4x slowdown - # (issue #792). Capping at len(uncached_work) keeps small jobs - # from spawning useless idle workers. - env_raw = os.environ.get("GRAPHIFY_MAX_WORKERS", "").strip() - env_cap = None - if env_raw: - try: - v = int(env_raw) - if v > 0: - env_cap = v - except ValueError: - pass - cpu_cap = env_cap if env_cap is not None else (os.cpu_count() or 4) - max_workers = min(cpu_cap, len(uncached_work)) - - # Windows ProcessPoolExecutor hard-caps at 61 workers (CPython limitation - # tied to WaitForMultipleObjects). Clamp here so every path — auto-compute, - # GRAPHIFY_MAX_WORKERS, and --max-workers — stays valid on >61-core boxes - # (issue #1298). Guard against 0 from an empty work list. - if sys.platform == "win32": - max_workers = min(max_workers, 61) - max_workers = max(max_workers, 1) + max_workers = resolve_max_workers(len(uncached_work), max_workers) # A one-worker pool buys no parallelism: it still pays process spawn plus an # IPC round trip per file, and it is the one residual case where the parent's @@ -5757,20 +5792,26 @@ def _extract_parallel( root_str = str(root) cache_loc_str = str(cache_location if cache_location is not None else root) work_items = [(idx, str(path), root_str, cache_loc_str) for idx, path in uncached_work] + batches = chunked(work_items, chunk_size_for(len(work_items), max_workers)) done_count = 0 failed: list[int] = [] # positions into uncached_work whose future failed _PROGRESS_INTERVAL = 100 try: with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as pool: - futures = { - pool.submit(_extract_single_file, item): pos - for pos, item in enumerate(work_items) - } + futures = {} + pos = 0 + for batch in batches: + fut = pool.submit(_extract_file_batch, batch) + futures[fut] = (pos, batch) + pos += len(batch) for future in concurrent.futures.as_completed(futures): + start, batch = futures[future] try: - idx, result = future.result() - per_file[idx] = result + results = future.result() + for idx, result in results: + per_file[idx] = result + done_count += len(results) except concurrent.futures.process.BrokenProcessPool: # #2444: a pool that dies while results are being consumed # raises BrokenProcessPool from every pending future. It @@ -5779,22 +5820,23 @@ def _extract_parallel( # swallowed here per-future — that left the remaining # per_file slots empty and silently dropped the files. raise - except Exception as exc: - pos = futures[future] + except Exception as extra: + extra_paths = f" (+{len(batch) - 1} more in batch)" if len(batch) > 1 else "" print( - f" warning: worker failed for {work_items[pos][1]}: {exc}", + f" warning: worker failed for {batch[0][1]}{extra_paths}: {extra}", file=sys.stderr, flush=True, ) - failed.append(pos) - done_count += 1 + failed.extend(range(start, start + len(batch))) + done_count += len(batch) if ( total_files >= _PROGRESS_INTERVAL and done_count % _PROGRESS_INTERVAL == 0 ): - print( + vprint( f" AST extraction: {done_count}/{len(uncached_work)} uncached files " f"({done_count * 100 // len(uncached_work)}%) [{max_workers} workers]", - flush=True, + file=sys.stdout, + prefix="", ) except concurrent.futures.process.BrokenProcessPool: # On Windows (spawn start method) the worker subprocesses re-import the @@ -5826,9 +5868,10 @@ def _extract_parallel( # corpus made the count jump upward at the end (cached hits + files with no # extractor never entered uncached_work), which read as inconsistent (#1693). _done = len(uncached_work) - print( + vprint( f" AST extraction: {_done}/{_done} uncached files (100%) [{max_workers} workers]", - flush=True, + file=sys.stdout, + prefix="", ) return True @@ -5848,9 +5891,10 @@ def _extract_sequential( and work_idx % _PROGRESS_INTERVAL == 0 and work_idx > 0 ): - print( + vprint( f" AST extraction: {work_idx}/{len(uncached_work)} uncached files ({work_idx * 100 // len(uncached_work)}%)", - flush=True, + file=sys.stdout, + prefix="", ) extractor = _get_extractor(path) if extractor is None: @@ -5866,10 +5910,10 @@ def _extract_sequential( if total_files >= _PROGRESS_INTERVAL: # Consistent denominator with the intermediate lines (#1693). _done = len(uncached_work) - print(f" AST extraction: {_done}/{_done} uncached files (100%)", flush=True) + vprint(f" AST extraction: {_done}/{_done} uncached files (100%)", file=sys.stdout, prefix="") -_PARALLEL_THRESHOLD = 20 +_PARALLEL_THRESHOLD = PARALLEL_THRESHOLD def extract( @@ -5901,8 +5945,10 @@ def extract( Anchors ids/source_file only as a fallback when `root` is unset. parallel: if True and there are >= _PARALLEL_THRESHOLD uncached files, use ProcessPoolExecutor for multi-core extraction. - max_workers: max subprocess count. Defaults to cpu_count (or the - value of GRAPHIFY_MAX_WORKERS if set), bounded by len(uncached_work). + max_workers: max subprocess/thread count for AST extract, cache + probe, and JS/Python symbol-resolution parses. Defaults to + cpu_count (or GRAPHIFY_MAX_WORKERS). The env/cpu default is + capped by the amount of work; an explicit value is not. resolution_context_nodes: read-only AST nodes from files that are NOT being extracted this run (an incremental rebuild's unchanged corpus, #2406). They extend the cross-file resolution indexes — @@ -5975,21 +6021,11 @@ def extract( cache_location = (cache_root if cache_root is not None else Path(".")).resolve() total = len(paths) - # Phase 1: separate cached hits from uncached work - per_file: list[dict | None] = [None] * total - uncached_work: list[tuple[int, Path]] = [] - - for i, path in enumerate(paths): - if _get_extractor(path) is None: - per_file[i] = {"nodes": [], "edges": []} - continue - bypass_cache = path.suffix in _JS_CACHE_BYPASS_SUFFIXES - if not bypass_cache: - cached = load_cached(path, root, cache_root=cache_location) - if cached is not None: - per_file[i] = cached - continue - uncached_work.append((i, path)) + # Phase 1: separate cached hits from uncached work (thread pool on large + # corpora — stat + JSON load is I/O bound). + per_file, uncached_work = _probe_cached_files( + paths, root, cache_location, parallel=parallel, max_workers=max_workers, + ) # Phase 2: extract uncached files (parallel or sequential) if uncached_work: @@ -6194,7 +6230,10 @@ def _describe_syntax_error(rel: str, line: "int | None", kept: int) -> str: # marker set in the per-file extractor. Populated just before the pass that uses it. callable_nids: set[str] = set() - _augment_symbol_resolution_edges(paths, all_nodes, all_edges, root) + _augment_symbol_resolution_edges( + paths, all_nodes, all_edges, root, + parallel=parallel, max_workers=max_workers, + ) # Merge a header-declared class (and its methods) with its sibling-impl # definition into ONE node (C/C++/ObjC #1547/#1556). Runs BEFORE the id-remap diff --git a/graphify/extractors/models.py b/graphify/extractors/models.py index 014907c0d..4fa0868b8 100644 --- a/graphify/extractors/models.py +++ b/graphify/extractors/models.py @@ -129,3 +129,14 @@ class _SymbolResolutionFacts: # is the binding introduced in the importing file: the alias when `from pkg # import submod as alias` is used, otherwise the submodule's own name (#2082). module_imports: list[tuple[Path, Path, int, str]] = field(default_factory=list) + + def extend(self, other: "_SymbolResolutionFacts") -> None: + """Append every fact list from *other*, preserving *other*'s order.""" + self.declarations.extend(other.declarations) + self.imports.extend(other.imports) + self.aliases.extend(other.aliases) + self.exports.extend(other.exports) + self.star_exports.extend(other.star_exports) + self.namespace_exports.extend(other.namespace_exports) + self.uses.extend(other.uses) + self.module_imports.extend(other.module_imports) diff --git a/graphify/extractors/resolution.py b/graphify/extractors/resolution.py index 44b027c69..c06dec04e 100644 --- a/graphify/extractors/resolution.py +++ b/graphify/extractors/resolution.py @@ -16,6 +16,9 @@ import re import sys +from graphify.parallel import map_in_process_pool +from graphify.progress import vprint + _TSCONFIG_ALIAS_CACHE: dict[str, dict[str, list[str]]] = {} @@ -1488,127 +1491,101 @@ def _ts_walk_class_members(class_node, source: bytes, path: Path, class_nid: str _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 +def _collect_js_facts_for_path(path: Path) -> _SymbolResolutionFacts: + """Parse one JS/TS/Vue/Svelte file and return its symbol-resolution facts. - trees: dict[Path, tuple[bytes, object]] = {} + Same walks as the historical all-files-then-next-pass loops, but scoped to + a single path so a process pool can parse many files at once. Merging + results in original path order reconstructs the serial fact lists. + """ + facts = _SymbolResolutionFacts() + parsed = _parse_js_tree(path) + if parsed is None: + return facts + source, root_node = 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) + ) - for path in js_paths: - resolved_path = path.resolve() - parsed = _parse_js_tree(path) - if parsed is None: + 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 - source, root_node = parsed - trees[resolved_path] = parsed + 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): - 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) - ) + 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) + ) - if node.type != "import_statement": - continue - raw_module = _js_module_specifier(node, source) - if raw_module is None: - continue + 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() - for imported_name, local_name in _js_named_specifiers(node, source, "import_specifier"): - facts.imports.append( - _SymbolImportFact( + namespace_name = _js_namespace_export_name(node, source) + if namespace_name is not None: + facts.namespace_exports.append( + _NamespaceExportFact( path, - local_name, + namespace_name, target_path, - imported_name, node.start_point[0] + 1, + type_only=stmt_type_only, ) ) - 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, + 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, ) ) - - 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( + for original_name, exported_name in _js_named_specifiers( export_clause, source, "export_specifier" ): facts.exports.append( @@ -1616,80 +1593,140 @@ def _collect_js_symbol_resolution_facts(paths: list[Path], facts: _SymbolResolut path, exported_name, node.start_point[0] + 1, - local_name=local_name, + target_path=target_path, + target_name=original_name, + type_only=stmt_type_only, ) ) - continue + continue - for exported_name in _js_exported_declaration_names(node, source): + 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=exported_name, + local_name=local_name, ) ) + continue - # `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 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, ) + ) - 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, - ) + # `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 - 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: + 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 - class_nid = _make_id(stem, class_name) - _ts_walk_class_members(node, source, path, class_nid, facts) + facts.uses.append( + _SymbolUseFact( + path, + source_id, + imported_name, + "calls", + "call", + node.start_point[0] + 1, + ) + ) + + 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) + return facts + + +def _collect_js_facts_worker(path_str: str) -> _SymbolResolutionFacts: + """Process-pool entry: path as str so the payload pickles cleanly.""" + return _collect_js_facts_for_path(Path(path_str)) + + +def _merge_fact_results( + facts: _SymbolResolutionFacts, + paths: list[Path], + parts: list[_SymbolResolutionFacts] | None, + collect_one, +) -> None: + """Extend *facts* in original *paths* order. + + ``parts is None`` means the pool declined (small job / one worker / BPP); + run *collect_one* sequentially. Otherwise *parts[i]* is the facts for + ``paths[i]``. + """ + if parts is None: + for path in paths: + facts.extend(collect_one(path)) + return + for part in parts: + facts.extend(part) + + +def _resolving_progress(msg: str) -> None: + vprint(msg, file=sys.stdout, prefix="") + + +def _collect_js_symbol_resolution_facts( + paths: list[Path], + facts: _SymbolResolutionFacts, + *, + parallel: bool = True, + max_workers: int | None = None, +) -> None: + js_paths = [ + path for path in paths + if path.suffix in _JS_CACHE_BYPASS_SUFFIXES + ] + if not js_paths: + return + _resolving_progress(f" resolving: JS/TS facts ({len(js_paths)} files)...") + parts = None + if parallel: + parts = map_in_process_pool( + _collect_js_facts_worker, + [str(p) for p in js_paths], + max_workers=max_workers, + ) + _merge_fact_results(facts, js_paths, parts, _collect_js_facts_for_path) + _resolving_progress(" resolving: JS/TS facts done") + def _parse_python_tree(path: Path): try: @@ -1824,92 +1861,123 @@ def _python_call_identifier(node, source: bytes) -> str | None: return _read_text(function_node, source) return None +def _collect_python_facts_for_path(path: Path, root: Path) -> _SymbolResolutionFacts: + """Parse one Python file into symbol-resolution facts (imports + uses).""" + facts = _SymbolResolutionFacts() + parsed = _parse_python_tree(path) + if parsed is None: + return facts + source, root_node = 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 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, + ) + ) + return facts + + +def _collect_python_facts_worker(args: tuple[str, str]) -> _SymbolResolutionFacts: + path_str, root_str = args + return _collect_python_facts_for_path(Path(path_str), Path(root_str)) + + def _collect_python_symbol_resolution_facts( paths: list[Path], root: Path, facts: _SymbolResolutionFacts, + *, + parallel: bool = True, + max_workers: int | None = None, ) -> None: py_paths = [path for path in paths if path.suffix == ".py"] if not py_paths: return + _resolving_progress(f" resolving: Python facts ({len(py_paths)} files)...") + parts = None + if parallel: + parts = map_in_process_pool( + _collect_python_facts_worker, + [(str(p), str(root)) for p in py_paths], + max_workers=max_workers, + ) + _merge_fact_results( + facts, + py_paths, + parts, + lambda p: _collect_python_facts_for_path(p, root), + ) + _resolving_progress(" resolving: Python facts done") - 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, + *, + parallel: bool = True, + max_workers: int | None = None, ) -> None: facts = _SymbolResolutionFacts() - _collect_js_symbol_resolution_facts(paths, facts) - _collect_python_symbol_resolution_facts(paths, root, facts) + _resolving_progress(f" resolving symbols across {len(paths)} files...") + _collect_js_symbol_resolution_facts( + paths, facts, parallel=parallel, max_workers=max_workers, + ) + _collect_python_symbol_resolution_facts( + paths, root, facts, parallel=parallel, max_workers=max_workers, + ) + _resolving_progress(" resolving: applying cross-file edges...") _apply_symbol_resolution_facts(paths, nodes, edges, root, facts) + _resolving_progress(" resolving: done") def _resolve_cross_file_imports( per_file: list[dict], diff --git a/graphify/hooks.py b/graphify/hooks.py index e535e1913..fb4f53e65 100644 --- a/graphify/hooks.py +++ b/graphify/hooks.py @@ -662,7 +662,10 @@ def _register_merge_driver(root: Path) -> str: try: for key, value in ( ("merge.graphify.name", "graphify graph.json union merge"), - ("merge.graphify.driver", driver), + # git config parses the argv with config-file syntax: `\` is an + # escape and an unquoted space ends the token. A quoted Windows + # path would otherwise store as `C:UsersFirst` on POSIX git (#2166). + ("merge.graphify.driver", driver.replace("\\", "\\\\").replace('"', '\\"')), ): _sp.run( ["git", "-C", str(root), "config", key, value], diff --git a/graphify/parallel.py b/graphify/parallel.py new file mode 100644 index 000000000..c08ba3f00 --- /dev/null +++ b/graphify/parallel.py @@ -0,0 +1,152 @@ +"""Process/thread-pool helpers for extraction and corpus scans. + +Worker counts honour ``GRAPHIFY_MAX_WORKERS`` the same way AST extraction does, +and Windows is clamped at 61 (CPython ``WaitForMultipleObjects`` limit). Callers +that cannot spawn a pool (one worker, small batches, ``BrokenProcessPool``) +fall back to in-process sequential work. +""" +from __future__ import annotations + +import os +import sys +from collections.abc import Callable, Sequence +from typing import TypeVar + +T = TypeVar("T") +R = TypeVar("R") + +# Same gate as extract._PARALLEL_THRESHOLD: below this a process pool's spawn +# cost dominates, and tests that mock ProcessPoolExecutor still exercise the +# sequential path on small fixtures. +PARALLEL_THRESHOLD = 20 + + +def resolve_max_workers(n_items: int, max_workers: int | None = None) -> int: + """Return a positive worker count for ``n_items`` pieces of work. + + ``max_workers`` (CLI ``--max-workers``) wins when given and is not + silently shrunk to ``n_items`` (extract() tests and the 1-file spawn + path pass an explicit count that must reach the pool). Otherwise + ``GRAPHIFY_MAX_WORKERS``, otherwise ``os.cpu_count()``, capped by + ``n_items`` so a 3-file job never starts 32 idle workers. + """ + if max_workers is not None: + # Caller-supplied count (CLI ``--max-workers``) is not silently shrunk + # to ``n_items``: extract() tests and the Windows 1-file spawn path + # pass an explicit value that must reach the pool. + workers = max(int(max_workers), 1) + else: + env_raw = os.environ.get("GRAPHIFY_MAX_WORKERS", "").strip() + env_cap = None + if env_raw: + try: + v = int(env_raw) + if v > 0: + env_cap = v + except ValueError: + pass + workers = env_cap if env_cap is not None else (os.cpu_count() or 4) + workers = min(max(int(workers), 1), n_items if n_items else 1) + if sys.platform == "win32": + workers = min(workers, 61) + return max(workers, 1) + + +def chunk_size_for(n_items: int, max_workers: int) -> int: + """Files (or items) per submitted task. + + Aims for about four chunks per worker so a slow file does not stall a + worker that already drained a giant one-shot batch, without submitting + one Future per file on a 50k-file corpus. + """ + if n_items <= 0 or max_workers <= 0: + return 1 + per_worker = max_workers * 4 + return max(1, min(32, (n_items + per_worker - 1) // per_worker)) + + +def chunked(items: Sequence[T], size: int) -> list[list[T]]: + size = max(int(size), 1) + return [list(items[i:i + size]) for i in range(0, len(items), size)] + + +def _run_batch(fn: Callable[[T], R], batch: list[T]) -> list[R]: + """Module-level so ProcessPoolExecutor can pickle it (Windows spawn).""" + return [fn(item) for item in batch] + + +def map_in_process_pool( + fn: Callable[[T], R], + items: Sequence[T], + *, + max_workers: int | None = None, + threshold: int = PARALLEL_THRESHOLD, +) -> list[R] | None: + """Map ``fn`` over ``items`` in a process pool, preserving order. + + Returns ``None`` when the caller should run sequentially instead: fewer + than ``threshold`` items, a resolved worker count of 1, or a + ``BrokenProcessPool`` (Windows spawn without an ``if __name__`` guard). + A non-BPP failure in one batch is retried in-process for that batch only. + """ + n = len(items) + if n == 0: + return [] + if n < threshold: + return None + workers = resolve_max_workers(n, max_workers) + if workers == 1: + return None + + import concurrent.futures + + batches = chunked(items, chunk_size_for(n, workers)) + results: list[R | None] = [None] * n + try: + with concurrent.futures.ProcessPoolExecutor(max_workers=workers) as pool: + futures: dict = {} + offset = 0 + for batch in batches: + fut = pool.submit(_run_batch, fn, batch) + futures[fut] = (offset, batch) + offset += len(batch) + for fut in concurrent.futures.as_completed(futures): + offset, batch = futures[fut] + try: + part = fut.result() + except concurrent.futures.process.BrokenProcessPool: + raise + except Exception: + part = _run_batch(fn, list(batch)) + results[offset:offset + len(part)] = part + except concurrent.futures.process.BrokenProcessPool: + return None + return results # type: ignore[return-value] + + +def map_in_thread_pool( + fn: Callable[[T], R], + items: Sequence[T], + *, + max_workers: int | None = None, + threshold: int = PARALLEL_THRESHOLD, +) -> list[R] | None: + """Map ``fn`` over ``items`` in a thread pool, preserving order. + + Same ``None``-means-run-sequentially contract as :func:`map_in_process_pool`. + Nested functions are fine here (threads, not pickling). + """ + n = len(items) + if n == 0: + return [] + if n < threshold: + return None + workers = resolve_max_workers(n, max_workers) + if workers == 1: + return None + + from concurrent.futures import ThreadPoolExecutor + + cs = chunk_size_for(n, workers) + with ThreadPoolExecutor(max_workers=workers) as pool: + return list(pool.map(fn, items, chunksize=cs)) diff --git a/graphify/progress.py b/graphify/progress.py new file mode 100644 index 000000000..ed2d676c8 --- /dev/null +++ b/graphify/progress.py @@ -0,0 +1,42 @@ +"""Opt-in progress chatter for detect/extract (``--verbose`` / ``GRAPHIFY_VERBOSE``). + +Quiet by default so a large corpus does not flood the terminal. Warnings and +errors still print from their call sites. ``--timing`` is independent. +""" +from __future__ import annotations + +import os +import sys +from typing import TextIO + +_verbose: bool | None = None + + +def set_verbose(on: bool | None) -> None: + """Force verbose on/off. ``None`` restores env-only (``GRAPHIFY_VERBOSE``).""" + global _verbose + _verbose = on + + +def verbose_enabled() -> bool: + if _verbose is True: + return True + if _verbose is False: + return False + return os.environ.get("GRAPHIFY_VERBOSE", "").strip().lower() in ( + "1", "true", "yes", "on", + ) + + +def vprint( + msg: str, + *, + file: TextIO | None = None, + prefix: str | None = "[graphify]", +) -> None: + """Print a progress line when verbose is on. Default stream is stderr.""" + if not verbose_enabled(): + return + dest = sys.stderr if file is None else file + text = f"{prefix} {msg}" if prefix else msg + print(text, file=dest, flush=True) diff --git a/pyproject.toml b/pyproject.toml index 7fd891dc8..31f10b21c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -145,9 +145,14 @@ graphify = ["skill.md", "skill-codex.md", "skill-opencode.md", "skill-kilo.md", [tool.pytest.ini_options] testpaths = ["tests"] norecursedirs = [ + # Pytest defaults must be restated: a custom list replaces, not extends, + # them. Omitting `.*` / `.hypothesis` makes Hypothesis warn at collection + # and walk its example database as if it were tests. + ".*", "*.egg", "_darcs", "build", "CVS", "dist", "node_modules", "venv", + "{arch}", ".hypothesis", "graphify-benchmark", "graphify_eval", "graphify_test", "worked", "llm-stack-corpus", "llm-stack-demo", "product-site", - "scripts", "ebook", ".github", "dist", "build", + "scripts", "ebook", ".github", ] [tool.bandit] diff --git a/tests/conftest.py b/tests/conftest.py index f581a7c1d..e48d8f412 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -61,7 +61,10 @@ def _sandbox_home(tmp_path_factory, monkeypatch): monkeypatch.setenv("LOCALAPPDATA", str(home / "AppData" / "Local")) monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) # escape hatch that bypasses Path.home monkeypatch.delenv("XDG_CONFIG_HOME", raising=False) + monkeypatch.delenv("GRAPHIFY_VERBOSE", raising=False) monkeypatch.setattr(Path, "home", classmethod(lambda cls: home)) + from graphify.progress import set_verbose + set_verbose(None) return home _ANALYZE_WARNING_FILTERS = ( diff --git a/tests/test_chunking.py b/tests/test_chunking.py index dea9f68f2..d453658be 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -311,10 +311,11 @@ def stray(chunk, **kwargs): } with patch("graphify.llm.extract_files_direct", side_effect=stray): - extract_corpus_parallel( - [a], backend="kimi", root=tmp_path, - token_budget=None, chunk_size=1, max_concurrency=1, - ) + with pytest.warns(RuntimeWarning, match="out-of-scope source_file 'B.py'"): + extract_corpus_parallel( + [a], backend="kimi", root=tmp_path, + token_budget=None, chunk_size=1, max_concurrency=1, + ) # B.py's cache is unchanged: the stray node was rejected, not merged in. after = load_cached(b, tmp_path, kind="semantic") @@ -464,10 +465,11 @@ def stray(chunk, **kwargs): } with patch("graphify.llm.extract_files_direct", side_effect=stray): - result = extract_corpus_parallel( - [a, c], backend="kimi", root=tmp_path, - token_budget=None, chunk_size=2, max_concurrency=1, - ) + with pytest.warns(RuntimeWarning, match="out-of-scope source_file 'B.py'"): + result = extract_corpus_parallel( + [a, c], backend="kimi", root=tmp_path, + token_budget=None, chunk_size=2, max_concurrency=1, + ) ids = {n["id"] for n in result["nodes"]} assert "b_stray" not in ids, "out-of-scope node leaked into the merged graph (#1895)" diff --git a/tests/test_detect.py b/tests/test_detect.py index 1bf6b056b..c96b65969 100644 --- a/tests/test_detect.py +++ b/tests/test_detect.py @@ -84,6 +84,18 @@ def test_detect_warns_small_corpus(): assert result["needs_graph"] is False assert result["warning"] is not None + +def test_detect_code_only_skips_word_count(tmp_path): + """--code-only must not open every file just to size the corpus.""" + (tmp_path / "app.py").write_text("x = 1\n", encoding="utf-8") + (tmp_path / "notes.md").write_text("# hello world " * 20 + "\n", encoding="utf-8") + result = detect(tmp_path, code_only=True) + assert result["total_words"] == 0 + assert any(f.endswith("app.py") for f in result["files"]["code"]) + assert any(f.endswith("notes.md") for f in result["files"]["document"]) + full = detect(tmp_path, code_only=False) + assert full["total_words"] > 0 + def test_detect_skips_noise_dot_dirs(): """Noise dot dirs (.next, .nuxt, .graphify cache, …) are skipped (#873). Non-noise dot dirs (.github, .claude, …) are now allowed through.""" @@ -456,8 +468,10 @@ def test_gitignore_keeps_tracked_file_but_drops_untracked_sibling(tmp_path): code = {Path(path).name for path in result["files"]["code"]} assert code == {"app.js", "fileWatcher.js"} - assert str(untracked) in result["ignored"] assert str(tracked) not in result["ignored"] + # Git's enumerator never visits gitignored untracked files, so scratch.js + # may be absent from ignored[] (the os.walk path still records it). + assert str(untracked) not in result["files"]["code"] def test_graphifyignore_still_excludes_git_tracked_file(tmp_path): @@ -473,7 +487,10 @@ def test_graphifyignore_still_excludes_git_tracked_file(tmp_path): result = detect(tmp_path) assert str(tracked) not in result["files"]["code"] - assert any(entry.rstrip(os.sep) == str(storage) for entry in result["ignored"]) + assert any( + entry.rstrip(os.sep) in (str(storage), str(tracked)) + for entry in result["ignored"] + ) def test_tracked_gitignore_exemption_works_for_subdirectory_scan(tmp_path): @@ -539,10 +556,12 @@ def _git_unavailable(*args, **kwargs): assert str(ignored_file) in result["ignored"] -def test_git_lsfiles_skipped_when_no_gitignore_contributes(tmp_path, monkeypatch): - """Optimization (#2759): a git repo with no .gitignore in play must not pay - the `git ls-files` subprocess — nothing can be gitignore-dropped, so the - tracked-exemption is moot. A .gitignore that DOES contribute still probes.""" +def test_git_lsfiles_enumerates_even_without_gitignore(tmp_path, monkeypatch): + """Git enumeration is the fast path on any Git repo, .gitignore or not. + + Previously ls-files ran only to exempt tracked files from gitignore; now + it replaces os.walk so ignored trees are never descended. + """ _git(tmp_path, "init", "-q") (tmp_path / "app.py").write_text("value = 1\n", encoding="utf-8") _git(tmp_path, "add", "app.py") @@ -557,17 +576,197 @@ def _spy(args, *a, **k): monkeypatch.setattr(detect_mod.subprocess, "run", _spy) - # No .gitignore anywhere -> gitignore contributes nothing -> no probe. - detect(tmp_path) - assert calls["ls_files"] == 0, "git ls-files ran despite no .gitignore in play" + result = detect(tmp_path) + assert calls["ls_files"] >= 1, "git ls-files must enumerate a Git working tree" + assert any(f.endswith("app.py") for f in result["files"]["code"]) - # Add a .gitignore -> gitignore now contributes -> probe happens (once). (tmp_path / ".gitignore").write_text("build/\n", encoding="utf-8") calls["ls_files"] = 0 detect(tmp_path) assert calls["ls_files"] >= 1, "git ls-files skipped even though .gitignore is present" +def test_git_enum_does_not_walk_gitignored_tree(tmp_path, monkeypatch): + """Ignored trees must not be os.walk'd — that is the million-file win.""" + _git(tmp_path, "init", "-q") + (tmp_path / "app.py").write_text("x = 1\n", encoding="utf-8") + _git(tmp_path, "add", "app.py") + (tmp_path / ".gitignore").write_text("blob/\n", encoding="utf-8") + blob = tmp_path / "blob" + blob.mkdir() + (blob / "noise.py").write_text("y = 2\n", encoding="utf-8") + + walked: list[str] = [] + real_walk = os.walk + + def _spy(start, *a, **k): + walked.append(os.path.abspath(start)) + return real_walk(start, *a, **k) + + monkeypatch.setattr(os, "walk", _spy) + + result = detect(tmp_path) + code = {Path(p).name for p in result["files"]["code"]} + assert code == {"app.py"} + blob_abs = os.path.abspath(blob) + assert not any( + w == blob_abs or w.startswith(blob_abs + os.sep) for w in walked + ), f"os.walk descended into gitignored blob/: {walked}" + + +def test_detect_skips_repo_tool_metadata_dir(tmp_path): + """Google repo-tool ``.repo/`` is object storage, not source (#playground).""" + (tmp_path / "app.py").write_text("x = 1\n", encoding="utf-8") + hidden = tmp_path / ".repo" / "projects" / "app.git" / "objects" + hidden.mkdir(parents=True) + (hidden / "pack.py").write_text("should_not_index = 1\n", encoding="utf-8") + + result = detect(tmp_path) + code = [Path(p).name for p in result["files"]["code"]] + assert "app.py" in code + assert "pack.py" not in code + assert any(f"{os.sep}.repo{os.sep}" in d or d.rstrip(os.sep).endswith(".repo") + for d in result["pruned_noise_dirs"]) + + +def test_detect_prunes_buildroot_output_trees(tmp_path): + """Buildroot ``output_/`` sysroots must not be walked.""" + (tmp_path / "main.c").write_text("int main(){return 0;}\n", encoding="utf-8") + out = tmp_path / "output_eq5_mips" + (out / "host").mkdir(parents=True) + (out / "target").mkdir() + (out / "images").mkdir() + (out / "host" / "usr.c").write_text("int x;\n", encoding="utf-8") + fake = tmp_path / "output_docs" + fake.mkdir() + (fake / "notes.py").write_text("x = 1\n", encoding="utf-8") + + result = detect(tmp_path) + code = {Path(p).name for p in result["files"]["code"]} + assert "main.c" in code + assert "notes.py" in code + assert "usr.c" not in code + + +def test_repo_workspace_enumerates_each_git_project(tmp_path, monkeypatch): + """A Google repo forest has no root .git; list each project instead of walking.""" + a = tmp_path / "proj_a" + b = tmp_path / "proj_b" + a.mkdir() + b.mkdir() + (a / "a.py").write_text("a = 1\n", encoding="utf-8") + (b / "b.py").write_text("b = 2\n", encoding="utf-8") + _git(a, "init", "-q") + _git(a, "add", "a.py") + _git(b, "init", "-q") + _git(b, "add", "b.py") + repo_dir = tmp_path / ".repo" + repo_dir.mkdir() + (repo_dir / "project.list").write_text("proj_a\nproj_b\n", encoding="utf-8") + junk = repo_dir / "projects" / "x.git" / "objects" + junk.mkdir(parents=True) + (junk / "blob.py").write_text("nope = 1\n", encoding="utf-8") + + walked: list[str] = [] + real_walk = os.walk + + def _spy(start, *a, **k): + walked.append(os.path.abspath(start)) + return real_walk(start, *a, **k) + + monkeypatch.setattr(os, "walk", _spy) + + result = detect(tmp_path) + names = {Path(p).name for p in result["files"]["code"]} + assert names == {"a.py", "b.py"} + assert "blob.py" not in names + junk_abs = os.path.abspath(junk) + assert not any( + w == junk_abs or w.startswith(junk_abs + os.sep) for w in walked + ) + + +def test_detect_enumerates_git_submodules(tmp_path, monkeypatch): + """A superproject with .gitmodules lists each submodule via git, not os.walk.""" + (tmp_path / "app.py").write_text("app = 1\n", encoding="utf-8") + lib = tmp_path / "vendor" / "lib" + lib.mkdir(parents=True) + (lib / "lib.py").write_text("lib = 1\n", encoding="utf-8") + (tmp_path / ".gitmodules").write_text( + '[submodule "vendor/lib"]\n\tpath = vendor/lib\n\turl = ./vendor/lib\n', + encoding="utf-8", + ) + _git(tmp_path, "init", "-q") + _git(tmp_path, "add", "app.py", ".gitmodules") + _git(lib, "init", "-q") + _git(lib, "add", "lib.py") + + walked: list[str] = [] + real_walk = os.walk + + def _spy(start, *a, **k): + walked.append(os.path.abspath(start)) + return real_walk(start, *a, **k) + + monkeypatch.setattr(os, "walk", _spy) + + result = detect(tmp_path) + names = {Path(p).name for p in result["files"]["code"]} + assert names == {"app.py", "lib.py"} + lib_abs = os.path.abspath(lib) + assert not any( + w == lib_abs or w.startswith(lib_abs + os.sep) for w in walked + ), f"os.walk descended into submodule: {walked}" + + +def test_detect_finds_nested_git_dirs_without_repo_or_submodules(tmp_path, monkeypatch): + """A directory of checkouts with no root .git and no .repo is enumerated per .git.""" + a = tmp_path / "alpha" + b = tmp_path / "beta" + a.mkdir() + b.mkdir() + (a / "a.py").write_text("a = 1\n", encoding="utf-8") + (b / "b.py").write_text("b = 2\n", encoding="utf-8") + _git(a, "init", "-q") + _git(a, "add", "a.py") + _git(b, "init", "-q") + _git(b, "add", "b.py") + + result = detect(tmp_path) + names = {Path(p).name for p in result["files"]["code"]} + assert names == {"a.py", "b.py"} + + +def test_detect_prefers_submodules_over_repo_manifest(tmp_path): + """Submodules win when both .gitmodules and .repo/project.list exist.""" + (tmp_path / "app.py").write_text("app = 1\n", encoding="utf-8") + lib = tmp_path / "lib" + lib.mkdir() + (lib / "lib.py").write_text("lib = 1\n", encoding="utf-8") + decoy = tmp_path / "decoy" + decoy.mkdir() + (decoy / "decoy.py").write_text("decoy = 1\n", encoding="utf-8") + (tmp_path / ".gitmodules").write_text( + '[submodule "lib"]\n\tpath = lib\n\turl = ./lib\n', + encoding="utf-8", + ) + _git(tmp_path, "init", "-q") + _git(tmp_path, "add", "app.py", ".gitmodules") + _git(lib, "init", "-q") + _git(lib, "add", "lib.py") + _git(decoy, "init", "-q") + _git(decoy, "add", "decoy.py") + repo_dir = tmp_path / ".repo" + repo_dir.mkdir() + (repo_dir / "project.list").write_text("decoy\n", encoding="utf-8") + + result = detect(tmp_path) + names = {Path(p).name for p in result["files"]["code"]} + assert "app.py" in names + assert "lib.py" in names + assert "decoy.py" not in names + + def test_gitignore_nested_below_root_prunes_whole_directory(tmp_path): """A nested .gitignore excluding a directory prevents descending into it.""" sub = tmp_path / "vendor" / "sub" @@ -684,6 +883,31 @@ def test_detect_skips_out_of_root_symlinked_directory_even_when_following(requir assert any("symlink target outside scan root" in item for item in result["skipped_sensitive"]) +def test_detect_rejects_regular_files_under_followed_out_of_root_memory_symlink( + requires_symlinks, tmp_path +): + """The memory walk skips ignore/noise pruning. A followed symlink directory + there lists the target's regular files, which are not themselves symlinks. + Admit must still reject them the way v8 rejected any resolve()-outside path. + """ + from graphify.paths import GRAPHIFY_OUT + + root = tmp_path / "root" + mem = root / GRAPHIFY_OUT / "memory" + mem.mkdir(parents=True) + (root / "ok.py").write_text("x = 1") + outside = tmp_path / "outside" + outside.mkdir() + (outside / "secret.py").write_text("token = 'leaked'") + (mem / "leak").symlink_to(outside) + + result = detect(root, follow_symlinks=True) + + assert not any("secret.py" in f for v in result["files"].values() for f in v) + assert any("ok.py" in f for f in result["files"]["code"]) + assert any("symlink target outside scan root" in item for item in result["skipped_sensitive"]) + + def test_detect_skips_out_of_root_symlinked_file_by_default(requires_symlinks, tmp_path): root = tmp_path / "root" root.mkdir() diff --git a/tests/test_extract.py b/tests/test_extract.py index 09d9b5c93..0e9aedb58 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -2200,10 +2200,10 @@ def __init__(self, *a, **kw): self._submitted = 0 def __enter__(self): return self def __exit__(self, *a): return False - def submit(self, fn, item): + def submit(self, fn, *args): self._submitted += 1 if self._submitted <= completed_before_break: - return GoodFuture(fn(item)) # extract in-process, eagerly + return GoodFuture(fn(*args)) # extract in-process, eagerly return BrokenFuture() monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", FakePool) @@ -2224,8 +2224,15 @@ def wrapped_sequential(uncached_work, *args, **kwargs): files = [FIXTURES / "sample.py"] * 25 result = extract_mod.extract(files, cache_root=tmp_path / "cache") + from graphify.parallel import chunk_size_for, chunked, resolve_max_workers + n = 25 + workers = resolve_max_workers(n, None) + batches = chunked(list(range(n)), chunk_size_for(n, workers)) + completed_idxs = [i for batch in batches[:completed_before_break] for i in batch] + leftover = [i for i in range(n) if i not in completed_idxs] + assert len(retried) == 1, "sequential fallback should have run exactly once" - assert sorted(retried[0]) == list(range(completed_before_break, 25)), ( + assert sorted(retried[0]) == leftover, ( "files whose futures completed before the pool broke must not be re-extracted" ) assert result["nodes"] @@ -2252,11 +2259,11 @@ def __init__(self, *a, **kw): self._submitted = 0 def __enter__(self): return self def __exit__(self, *a): return False - def submit(self, fn, item): + def submit(self, fn, *args): self._submitted += 1 if self._submitted == 1: return FailingFuture() - return GoodFuture(fn(item)) + return GoodFuture(fn(*args)) monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", FakePool) monkeypatch.setattr( @@ -2276,7 +2283,12 @@ def wrapped_sequential(uncached_work, *args, **kwargs): files = [FIXTURES / "sample.py"] * 25 result = extract_mod.extract(files, cache_root=tmp_path / "cache") - assert retried == [[0]], "only the failed file may be retried, exactly once" + from graphify.parallel import chunk_size_for, chunked, resolve_max_workers + first_batch = chunked( + list(range(25)), + chunk_size_for(25, resolve_max_workers(25, None)), + )[0] + assert retried == [first_batch], "only the failed batch may be retried, exactly once" assert result["nodes"] err = capsys.readouterr().err assert "worker failed" in err @@ -2313,11 +2325,11 @@ def __init__(self, *a, **kw): self._submitted = 0 def __enter__(self): return self def __exit__(self, *a): return False - def submit(self, fn, item): + def submit(self, fn, *args): self._submitted += 1 if self._submitted == 1: # boom.go is first in the batch return FailingFuture() - return GoodFuture(fn(item)) + return GoodFuture(fn(*args)) monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", FakePool) monkeypatch.setattr( @@ -2339,14 +2351,71 @@ def wrapped_sequential(uncached_work, per_file, *args, **kwargs): files = [bad_file] + [FIXTURES / "sample.py"] * 24 result = extract_mod.extract(files, cache_root=tmp_path / "cache") + from graphify.parallel import chunk_size_for, chunked, resolve_max_workers + first_batch = chunked( + list(range(25)), + chunk_size_for(25, resolve_max_workers(25, None)), + )[0] assert captured["calls"] == 1, "the retry must be bounded: one pass, no loop" - assert captured["retry_indices"] == [0] + assert captured["retry_indices"] == first_batch + assert 0 in captured["retry_indices"] assert "error" in captured["per_file"][0], ( "a twice-failing file must carry an error marker, not a clean empty" ) assert result["nodes"], "the other files must still complete" +def test_extract_parallel_submits_file_batches(tmp_path, monkeypatch): + """One Future per file on a large corpus is too much pickle/scheduling; + _extract_parallel must submit chunks (and still extract every file).""" + import concurrent.futures + from graphify import extract as extract_mod + from graphify.parallel import chunk_size_for, chunked, resolve_max_workers + + class GoodFuture: + def __init__(self, value): + self._value = value + + def result(self): + return self._value + + submitted: list[int] = [] + + class FakePool: + def __init__(self, *a, **kw): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def submit(self, fn, *args): + # Symbol-resolution also opens a ProcessPoolExecutor; only count + # AST extraction batches (one positional payload). + if fn is extract_mod._extract_file_batch: + submitted.append(len(args[0])) + return GoodFuture(fn(*args)) + + monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", FakePool) + monkeypatch.setattr( + concurrent.futures, "as_completed", lambda futures: iter(futures) + ) + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + + files = [FIXTURES / "sample.py"] * 25 + result = extract_mod.extract(files, cache_root=tmp_path / "cache") + + expected = chunked( + list(range(25)), + chunk_size_for(25, resolve_max_workers(25, None)), + ) + assert submitted == [len(b) for b in expected] + assert len(submitted) < 25 + assert result["nodes"] + + def test_extract_legitimately_empty_result_keeps_no_error_marker( tmp_path, monkeypatch, capsys ): @@ -3805,13 +3874,19 @@ def test_extract_progress_final_line_uses_consistent_denominator(tmp_path, capsy # #1693: intermediate progress lines count against uncached_work; the final # "100%" line must NOT switch to total_files (which includes cached hits and # files with no extractor), or the count appears to jump upward at the end. + from graphify.progress import set_verbose + for i in range(100): (tmp_path / f"m{i}.py").write_text(f"def f{i}():\n return {i}\n") for i in range(5): (tmp_path / f"s{i}.r").write_text(f"g{i} <- function(x) x\n") # no extractor paths = sorted(tmp_path.glob("*.py")) + sorted(tmp_path.glob("*.r")) # total 105 - extract(paths, cache_root=tmp_path, parallel=False) + set_verbose(True) + try: + extract(paths, cache_root=tmp_path, parallel=False) + finally: + set_verbose(None) out = capsys.readouterr().out # final progress line reports the uncached count (100), not the total (105) @@ -3819,6 +3894,38 @@ def test_extract_progress_final_line_uses_consistent_denominator(tmp_path, capsy assert "105/105 files" not in out, "final line must not switch to total_files (#1693)" +def test_extract_resolving_progress_is_verbose_only(tmp_path, capsys): + from graphify.progress import set_verbose + + a = tmp_path / "a.py" + b = tmp_path / "b.py" + a.write_text("def a():\n return 1\n") + b.write_text("from a import a\ndef b():\n return a()\n") + extract([a, b], cache_root=tmp_path, parallel=False) + assert "resolving symbols" not in capsys.readouterr().out + + set_verbose(True) + try: + extract([a, b], cache_root=tmp_path, parallel=False) + finally: + set_verbose(None) + out = capsys.readouterr().out + assert "resolving symbols across 2 files" in out + assert "resolving: Python facts (2 files)" in out + assert "resolving: applying cross-file edges" in out + assert "resolving: done" in out + + +def test_extract_resolving_progress_via_env(tmp_path, capsys, monkeypatch): + monkeypatch.setenv("GRAPHIFY_VERBOSE", "1") + a = tmp_path / "a.py" + a.write_text("def a():\n return 1\n") + extract([a], cache_root=tmp_path, parallel=False) + out = capsys.readouterr().out + assert "resolving symbols across 1 files" in out + assert "resolving: Python facts (1 files)" in out + + def test_get_extractor_routes_matlab_m_away_from_objc(tmp_path): # #1702: .m is shared by Objective-C and MATLAB. A real ObjC .m still routes to # extract_objc, but a MATLAB .m must NOT be force-parsed by the ObjC grammar diff --git a/tests/test_extract_cli.py b/tests/test_extract_cli.py index 4c9fb445f..80d400d81 100644 --- a/tests/test_extract_cli.py +++ b/tests/test_extract_cli.py @@ -886,6 +886,45 @@ def test_extract_timing_flag_emits_stage_timings(monkeypatch, tmp_path, capsys): assert "graphify timing" not in capsys.readouterr().err +def test_extract_verbose_flag_emits_detect_and_resolving_progress(monkeypatch, tmp_path, capsys): + """Quiet by default; --verbose turns on detect/AST/resolving chatter.""" + monkeypatch.setattr(mainmod, "_check_skill_version", lambda _: None) + code = tmp_path / "src" + code.mkdir() + (code / "a.py").write_text("def a():\n return 1\n") + (code / "b.py").write_text("from a import a\ndef b():\n return a()\n") + + monkeypatch.setattr( + mainmod.sys, "argv", + ["graphify", "extract", str(code), "--code-only", "--no-cluster", + "--out", str(tmp_path / "quiet")], + ) + with pytest.raises(SystemExit) as exc: + mainmod.main() + assert exc.value.code == 0 + captured = capsys.readouterr() + combined = captured.out + captured.err + assert "[graphify] scanning" not in combined + assert "resolving symbols" not in combined + assert "AST extraction:" not in combined + assert "wrote" in captured.out + + monkeypatch.setattr( + mainmod.sys, "argv", + ["graphify", "extract", str(code), "--code-only", "--no-cluster", + "--out", str(tmp_path / "loud"), "--verbose"], + ) + with pytest.raises(SystemExit) as exc2: + mainmod.main() + assert exc2.value.code == 0 + captured2 = capsys.readouterr() + combined2 = captured2.out + captured2.err + assert "[graphify] scanning" in combined2 or "[graphify extract] scanning" in combined2 + assert "resolving symbols across" in combined2 + assert "resolving: Python facts" in combined2 + assert "resolving: done" in combined2 + + @pytest.mark.parametrize( "postgres_args", [["--postgres", "test-dsn"], ["--postgres=test-dsn"]], @@ -1238,7 +1277,7 @@ def test_failed_extra_is_retried_and_recovers(monkeypatch, tmp_path, capsys): monkeypatch.setitem(extractmod._DISPATCH, ".sql", _ok_sql) _run_extract(monkeypatch, argv) out_text = capsys.readouterr().out - assert "1 code" in out_text, f"schema.sql must be in the changed set: {out_text}" + assert "1 re-extracted" in out_text, f"schema.sql must be in the changed set: {out_text}" assert any("schema.sql" in s for s in _node_sources(graph_path)), ( "recovered schema.sql must contribute nodes to graph.json" ) @@ -1274,10 +1313,9 @@ def test_permanent_failure_does_not_wedge(monkeypatch, tmp_path, capsys): for run in (2, 3): _run_extract(monkeypatch, argv) out_text = capsys.readouterr().out - assert "1 code" in out_text, ( + assert "1 re-extracted" in out_text, ( f"run {run}: the failed file must be retried, not frozen: {out_text}" ) - assert "1 re-extracted" in out_text, f"run {run}: {out_text}" sources = _node_sources(graph_path) assert any("keep.py" in s for s in sources), f"run {run}: graph must stay stable" assert not any("schema.sql" in s for s in sources) diff --git a/tests/test_extract_code_only_cli.py b/tests/test_extract_code_only_cli.py index 2776b2aa4..3e817afe4 100644 --- a/tests/test_extract_code_only_cli.py +++ b/tests/test_extract_code_only_cli.py @@ -69,6 +69,15 @@ def test_extract_usage_advertises_code_only(tmp_path): ) +def test_extract_usage_advertises_verbose(tmp_path): + r = subprocess.run( + [PYTHON, "-m", "graphify", "extract"], + cwd=tmp_path, capture_output=True, text=True, + ) + assert r.returncode != 0 + assert "--verbose" in r.stdout + r.stderr + + def _run_relative_out(repo: Path, *extra: str): """Like _run but with a RELATIVE GRAPHIFY_OUT so --out/--output controls the parent dir (an absolute GRAPHIFY_OUT would override the flag).""" diff --git a/tests/test_parallel.py b/tests/test_parallel.py new file mode 100644 index 000000000..8a7d0e0be --- /dev/null +++ b/tests/test_parallel.py @@ -0,0 +1,246 @@ +"""Helpers in graphify.parallel and end-to-end parallel vs sequential identity.""" +from __future__ import annotations + +from pathlib import Path + +from graphify.parallel import ( + PARALLEL_THRESHOLD, + chunk_size_for, + chunked, + map_in_process_pool, + map_in_thread_pool, + resolve_max_workers, +) + + +def _double(x: int) -> int: + """Module-level so a spawn-start process pool can pickle it.""" + return x * 2 + + +def _fingerprint(result: dict) -> tuple[tuple[str, ...], tuple[tuple[str, str, str, str], ...]]: + nodes = tuple(sorted(str(n["id"]) for n in result["nodes"])) + edges = tuple(sorted( + ( + str(e["source"]), + str(e["target"]), + str(e["relation"]), + str(e.get("context") or ""), + ) + for e in result["edges"] + )) + return nodes, edges + + +def test_chunk_size_for_aims_for_four_chunks_per_worker(): + assert chunk_size_for(25, 2) == 4 + assert chunk_size_for(1, 8) == 1 + assert chunk_size_for(10_000, 8) == 32 + assert chunk_size_for(0, 4) == 1 + + +def test_chunked_splits_and_keeps_a_short_tail(): + assert chunked([1, 2, 3, 4, 5], 2) == [[1, 2], [3, 4], [5]] + assert chunked([], 4) == [] + + +def test_resolve_max_workers_env_capped_by_item_count(monkeypatch): + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "8") + assert resolve_max_workers(3, None) == 3 + assert resolve_max_workers(25, None) == 8 + + +def test_resolve_max_workers_explicit_is_not_capped_by_item_count(monkeypatch): + monkeypatch.delenv("GRAPHIFY_MAX_WORKERS", raising=False) + assert resolve_max_workers(1, 2) == 2 + + +def test_map_in_process_pool_declines_small_jobs(): + assert map_in_process_pool(_double, list(range(5))) is None + + +def test_map_in_process_pool_declines_one_worker(monkeypatch): + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "1") + assert map_in_process_pool(_double, list(range(PARALLEL_THRESHOLD + 5))) is None + + +def test_map_in_process_pool_preserves_order(monkeypatch): + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + items = list(range(PARALLEL_THRESHOLD + 5)) + got = map_in_process_pool(_double, items) + assert got == [_double(x) for x in items] + + +def test_map_in_process_pool_bpp_returns_none(monkeypatch): + from concurrent.futures.process import BrokenProcessPool + import concurrent.futures + + class FakePool: + def __init__(self, *a, **kw): + pass + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def submit(self, *a, **kw): + raise BrokenProcessPool("simulated spawn failure") + + monkeypatch.setattr(concurrent.futures, "ProcessPoolExecutor", FakePool) + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + assert map_in_process_pool(_double, list(range(25))) is None + + +def test_map_in_thread_pool_preserves_order(monkeypatch): + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + items = list(range(PARALLEL_THRESHOLD + 5)) + got = map_in_thread_pool(_double, items) + assert got == [_double(x) for x in items] + + +def test_detect_parallel_word_count_matches_sequential(tmp_path, monkeypatch): + from graphify import detect as det + + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + src = tmp_path / "src" + src.mkdir() + for i in range(25): + (src / f"doc{i}.md").write_text(f"alpha beta gamma {i}\n" * 3) + + real = det.map_in_thread_pool + used_pool = {"yes": False} + + def tracking(*args, **kwargs): + result = real(*args, **kwargs) + if result is not None: + used_pool["yes"] = True + return result + + monkeypatch.setattr(det, "map_in_thread_pool", lambda *a, **k: None) + sequential = det.detect(src, cache_root=tmp_path / "c_seq") + monkeypatch.setattr(det, "map_in_thread_pool", tracking) + parallel = det.detect(src, cache_root=tmp_path / "c_par") + + assert used_pool["yes"], "25 files must take the thread-pool word-count path" + assert sequential["total_words"] == parallel["total_words"] + assert sequential["files"] == parallel["files"] + assert sequential["total_files"] == 25 + + +def test_detect_parallel_walk_matches_sequential(tmp_path, monkeypatch): + """Sibling trees + nested gitignores must match a sequential walk.""" + from graphify import detect as det + + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "4") + src = tmp_path / "src" + src.mkdir() + for i in range(25): + d = src / f"pkg{i}" + d.mkdir() + (d / ".gitignore").write_text("*.log\n", encoding="utf-8") + (d / f"mod{i}.py").write_text(f"x = {i}\n", encoding="utf-8") + (d / f"noise{i}.log").write_text("nope\n", encoding="utf-8") + + real = det.map_in_thread_pool + monkeypatch.setattr(det, "map_in_thread_pool", lambda *a, **k: None) + sequential = det.detect(src, cache_root=tmp_path / "c_seq") + monkeypatch.setattr(det, "map_in_thread_pool", real) + parallel = det.detect(src, cache_root=tmp_path / "c_par") + + assert sequential["files"] == parallel["files"] + assert sequential["ignored"] == parallel["ignored"] + assert sequential["unclassified"] == parallel["unclassified"] + assert sequential["total_words"] == parallel["total_words"] + assert sequential["total_files"] == 25 + assert sequential["graphifyignore_patterns"] == parallel["graphifyignore_patterns"] + + +def test_extract_python_parallel_matches_sequential(tmp_path, monkeypatch): + from graphify.extract import extract + + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + pkg = tmp_path / "pkg" + pkg.mkdir() + (pkg / "__init__.py").write_text("", encoding="utf-8") + paths = [pkg / "__init__.py"] + for i in range(25): + p = pkg / f"m{i}.py" + if i == 0: + p.write_text("def f0():\n return 0\n", encoding="utf-8") + else: + p.write_text( + f"from .m{i - 1} import f{i - 1}\n\n" + f"def f{i}():\n return f{i - 1}()\n", + encoding="utf-8", + ) + paths.append(p) + + sequential = extract( + paths, cache_root=tmp_path / "seq", root=tmp_path, parallel=False, + ) + parallel = extract( + paths, cache_root=tmp_path / "par", root=tmp_path, + parallel=True, max_workers=2, + ) + assert _fingerprint(parallel) == _fingerprint(sequential) + assert any(e["relation"] == "imports" for e in sequential["edges"]) + + +def test_extract_js_parallel_matches_sequential(tmp_path, monkeypatch): + from graphify.extract import extract + + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + paths: list[Path] = [] + for i in range(25): + p = tmp_path / f"m{i}.js" + if i == 0: + p.write_text("export function f0() { return 0 }\n", encoding="utf-8") + else: + p.write_text( + f"import {{ f{i - 1} }} from './m{i - 1}.js'\n" + f"export function f{i}() {{ return f{i - 1}() }}\n", + encoding="utf-8", + ) + paths.append(p) + + sequential = extract( + paths, cache_root=tmp_path / "seq", root=tmp_path, parallel=False, + ) + parallel = extract( + paths, cache_root=tmp_path / "par", root=tmp_path, + parallel=True, max_workers=2, + ) + assert _fingerprint(parallel) == _fingerprint(sequential) + assert any(e["relation"] == "imports" for e in sequential["edges"]) + + +def test_extract_warm_cache_skips_ast_pool(tmp_path, monkeypatch): + from graphify import extract as extract_mod + + monkeypatch.setenv("GRAPHIFY_MAX_WORKERS", "2") + paths = [] + for i in range(25): + p = tmp_path / f"m{i}.py" + p.write_text(f"def f{i}():\n return {i}\n", encoding="utf-8") + paths.append(p) + + extract_mod.extract( + paths, cache_root=tmp_path / "c", root=tmp_path, + parallel=True, max_workers=2, + ) + + spawned = {"n": 0} + real = extract_mod._extract_parallel + + def wrapped(*args, **kwargs): + spawned["n"] += 1 + return real(*args, **kwargs) + + monkeypatch.setattr(extract_mod, "_extract_parallel", wrapped) + extract_mod.extract( + paths, cache_root=tmp_path / "c", root=tmp_path, + parallel=True, max_workers=2, + ) + assert spawned["n"] == 0, "a warm AST cache must not re-enter the process pool"