diff --git a/CHANGELOG.md b/CHANGELOG.md index d336d5e..5d59365 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,14 @@ ## Unreleased +### Added + +- **rg-backed `backlinks`**. `okf-graph.py backlinks` uses ripgrep (when on + PATH, or `OKF_RG_PATH` / `PKC_RG_PATH`) to find inbound files and parse only + those hits. Ranking/identity matches a full `load_bundle`. `--no-rg` forces + the previous scan. Ambiguous title/stem queries still load the bundle so they + keep erroring instead of silently picking a file. + ## 0.8.1 — 2026-08-24 - Noun-ownership migration guide for existing second brains: diff --git a/scripts/okf-graph.py b/scripts/okf-graph.py index 8ca49b6..b700a29 100755 --- a/scripts/okf-graph.py +++ b/scripts/okf-graph.py @@ -18,7 +18,10 @@ import argparse import html import json +import os import re +import shutil +import subprocess import sys from collections import defaultdict, deque from dataclasses import dataclass, field @@ -35,6 +38,54 @@ load_default_registry = None # type: ignore +def find_rg() -> str | None: + for var in ("OKF_RG_PATH", "PKC_RG_PATH", "SECOND_BRAIN_RG_PATH"): + override = (os.environ.get(var) or "").strip() + if not override: + continue + p = Path(override) + if p.is_file() and os.access(p, os.X_OK): + return str(p.resolve()) + found = shutil.which(override) + if found: + return found + return shutil.which("rg") + + +def rg_list_files( + root: Path, + pattern: str, + *, + fixed_string: bool = True, + ignore_case: bool = False, + timeout: float = 30.0, +) -> list[Path] | None: + """`rg -l` for one pattern. None = rg missing/failed; [] = no hits.""" + rg = find_rg() + if not rg or not pattern: + return None + cmd = [rg, "-l", "--no-messages", "--color", "never", "--glob", "*.md"] + if ignore_case: + cmd.append("-i") + if fixed_string: + cmd.append("-F") + cmd.extend(["--", pattern, str(root)]) + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False) + except (OSError, subprocess.TimeoutExpired): + return None + if proc.returncode not in (0, 1): + return None + files: list[Path] = [] + for line in proc.stdout.splitlines(): + line = line.strip() + if not line: + continue + p = Path(line) + files.append(p.resolve() if p.is_absolute() else (root / p).resolve()) + return files + + FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) # The label alternation keeps `[^\]]` — the entirety of the previous pattern's # label language — and *adds* balanced `[...]` pairs, tried first. Keeping the @@ -758,12 +809,38 @@ def cmd_impact(bundle: Path, concept: str) -> int: return 0 -def cmd_backlinks(bundle: Path, concept: str) -> int: +def cmd_backlinks(bundle: Path, concept: str, *, use_rg: bool | None = None) -> int: + fast = _existing_rel(bundle, concept) + if use_rg is not False and fast and find_rg(): + payload = _backlinks_via_rg(bundle, fast) + if payload is not None: + print(json.dumps(payload, indent=2)) + return 0 concepts = load_bundle(bundle) inbound_map = build_inbound(concepts) target = resolve_or_error(concepts, concept) if not target: return 1 + print(json.dumps(_backlinks_from_graph(concepts, inbound_map, target), indent=2)) + return 0 + + +def _existing_rel(bundle: Path, concept: str) -> str | None: + """Return a bundle-relative path only when `concept` already names a file. + + Title/stem lookup stays on the full load so ambiguous `page.md` queries + still error instead of silently picking one file. + """ + q = concept.strip().lstrip("/") + cand = bundle / q + if cand.is_file(): + return cand.relative_to(bundle).as_posix() + return None + + +def _backlinks_from_graph( + concepts: dict[str, Concept], inbound_map: dict[str, list[str]], target: str +) -> dict[str, Any]: bl = [] for i in inbound_map.get(target, []): rels = [e.rel for e in concepts[i].edges if e.target == target] @@ -775,8 +852,54 @@ def cmd_backlinks(bundle: Path, concept: str) -> int: "rels": rels or ["links_to"], } ) - print(json.dumps({"target": target, "backlinks": bl}, indent=2)) - return 0 + return {"target": target, "backlinks": bl, "engine": "scan"} + + +def _backlinks_via_rg(bundle: Path, target_rel: str) -> dict[str, Any] | None: + """Parse only files that mention the target path. None = fall back.""" + needles = ["/" + target_rel, target_rel, Path(target_rel).name] + found: set[Path] = set() + any_run = False + for needle in needles: + hits = rg_list_files(bundle, needle, fixed_string=True, ignore_case=False) + if hits is None: + continue + any_run = True + found.update(hits) + if not any_run: + return None + bl: list[dict[str, Any]] = [] + seen: set[str] = set() + target_path = (bundle / target_rel).resolve() + for path in sorted(found): + if path.resolve() == target_path: + continue + try: + rel = path.relative_to(bundle).as_posix() + except ValueError: + continue + if any(part.startswith(".") for part in Path(rel).parts): + continue + text = path.read_text(encoding="utf-8", errors="replace") + meta = parse_frontmatter(text) + md_edges = extract_markdown_links(text, path, bundle) + fm_edges = extract_frontmatter_links(meta, path, bundle) + edges = merge_edges(md_edges, fm_edges) + rels = [e.rel for e in edges if e.target == target_rel] + if not rels: + continue + if rel in seen: + continue + seen.add(rel) + bl.append( + { + "id": rel, + "title": str(meta.get("title") or path.stem), + "type": str(meta.get("type") or ("Index" if path.name == "index.md" else "Unknown")), + "rels": rels, + } + ) + return {"target": target_rel, "backlinks": bl, "engine": "rg"} def cmd_subgraph(bundle: Path, concept: str, hops: int) -> int: @@ -1212,6 +1335,17 @@ def main() -> int: s = sub.add_parser(name) s.add_argument("bundle") s.add_argument("concept") + if name == "backlinks": + s.add_argument( + "--rg", + action="store_true", + help="Use ripgrep to find inbound files (default when rg is on PATH)", + ) + s.add_argument( + "--no-rg", + action="store_true", + help="Disable ripgrep; load the whole bundle", + ) s = sub.add_parser("subgraph") s.add_argument("bundle") @@ -1280,7 +1414,17 @@ def main() -> int: if args.cmd == "impact": return cmd_impact(bundle, args.concept) if args.cmd == "backlinks": - return cmd_backlinks(bundle, args.concept) + if getattr(args, "rg", False) and getattr(args, "no_rg", False): + print(json.dumps({"error": "--rg and --no-rg are mutually exclusive"})) + return 2 + use_rg: bool | None + if getattr(args, "no_rg", False): + use_rg = False + elif getattr(args, "rg", False): + use_rg = True + else: + use_rg = None + return cmd_backlinks(bundle, args.concept, use_rg=use_rg) if args.cmd == "subgraph": return cmd_subgraph(bundle, args.concept, args.hops) if args.cmd == "pack": diff --git a/tests/fixtures/fake_rg.py b/tests/fixtures/fake_rg.py new file mode 100755 index 0000000..b64c9bd --- /dev/null +++ b/tests/fixtures/fake_rg.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Minimal `rg -l` stand-in for tests. Not a real ripgrep. + +Understands: + -l / --files-with-matches + -i / --ignore-case + -F / --fixed-strings + --glob GLOB (including !negations) + --no-messages --color never + pattern PATH +""" + +from __future__ import annotations + +import argparse +import fnmatch +import os +import re +import sys +from pathlib import Path + + +def match_globs(rel: str, globs: list[str]) -> bool: + include = [g for g in globs if not g.startswith("!")] + exclude = [g[1:] for g in globs if g.startswith("!")] + ok = True + if include: + ok = any(fnmatch.fnmatch(rel, g) or fnmatch.fnmatch(Path(rel).name, g) for g in include) + for g in exclude: + if fnmatch.fnmatch(rel, g) or fnmatch.fnmatch(Path(rel).name, g): + return False + return ok + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(add_help=False) + p.add_argument("-l", "--files-with-matches", action="store_true") + p.add_argument("-i", "--ignore-case", action="store_true") + p.add_argument("-F", "--fixed-strings", action="store_true") + p.add_argument("--glob", action="append", default=[]) + p.add_argument("--no-messages", action="store_true") + p.add_argument("--color", default="never") + p.add_argument("pattern") + p.add_argument("path", nargs="?", default=".") + args = p.parse_args(argv) + + root = Path(args.path).resolve() + flags = re.I if args.ignore_case else 0 + if args.fixed_strings: + needle = args.pattern.lower() if args.ignore_case else args.pattern + pred = lambda text: needle in (text.lower() if args.ignore_case else text) + else: + try: + rx = re.compile(args.pattern, flags) + except re.error: + return 2 + pred = lambda text: rx.search(text) is not None + + hits = 0 + if root.is_file(): + files = [root] + base = root.parent + else: + files = sorted(root.rglob("*")) + base = root + for path in files: + if not path.is_file(): + continue + try: + rel = path.relative_to(base).as_posix() + except ValueError: + rel = path.name + if args.glob and not match_globs(rel, args.glob): + continue + try: + text = path.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + if pred(text): + print(path) + hits += 1 + return 0 if hits else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_okf_graph.py b/tests/test_okf_graph.py index ac7817a..cf382bc 100644 --- a/tests/test_okf_graph.py +++ b/tests/test_okf_graph.py @@ -12,6 +12,7 @@ import importlib.util import json +import os import re import subprocess import sys @@ -716,6 +717,65 @@ def test_known_rels_covers_sibling_plugin_vocabularies(): assert rels <= g.KNOWN_RELS, f"{name} is not a subset of KNOWN_RELS" +FAKE_RG = REPO / "tests" / "fixtures" / "fake_rg.py" + + +def test_backlinks_rg_matches_scan(): + """rg prefilter must not change backlink identity vs a full load.""" + FAKE_RG.chmod(0o755) + env = os.environ.copy() + env["OKF_RG_PATH"] = str(FAKE_RG) + target = "knowledge/okf-conventions.md" + scan = subprocess.run( + [sys.executable, str(SCRIPT), "backlinks", "sample-okf", target, "--no-rg"], + capture_output=True, + text=True, + cwd=REPO, + env=env, + ) + accel = subprocess.run( + [sys.executable, str(SCRIPT), "backlinks", "sample-okf", target, "--rg"], + capture_output=True, + text=True, + cwd=REPO, + env=env, + ) + assert scan.returncode == 0, scan.stderr + assert accel.returncode == 0, accel.stderr + s = json.loads(scan.stdout) + a = json.loads(accel.stdout) + assert s["engine"] == "scan", s + assert a["engine"] == "rg", a + assert s["target"] == a["target"] == target + sid = sorted(b["id"] for b in s["backlinks"]) + aid = sorted(b["id"] for b in a["backlinks"]) + assert sid == aid, (sid, aid) + + +def test_backlinks_ambiguous_still_errors_with_rg(): + """Path-fast-path must not swallow an ambiguous stem query.""" + FAKE_RG.chmod(0o755) + env = os.environ.copy() + env["OKF_RG_PATH"] = str(FAKE_RG) + with tempfile.TemporaryDirectory() as td: + bundle = Path(td) / "b" + (bundle / "a").mkdir(parents=True) + (bundle / "b").mkdir() + (bundle / "index.md").write_text("---\ntitle: Root\n---\n[a](/a/page.md) [b](/b/page.md)\n") + (bundle / "a" / "page.md").write_text("---\ntitle: A\ntype: Reference\n---\n") + (bundle / "b" / "page.md").write_text("---\ntitle: B\ntype: Reference\n---\n") + proc = subprocess.run( + [sys.executable, str(SCRIPT), "backlinks", str(bundle), "page.md", "--rg"], + capture_output=True, + text=True, + cwd=REPO, + env=env, + ) + assert proc.returncode == 1, proc.stdout + out = json.loads(proc.stdout) + assert "error" in out and out.get("candidates") == ["a/page.md", "b/page.md"], out + + def main() -> int: quiet = "-q" in sys.argv tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]