wip:RAG pipeline:general structure #6 - ingestion: extract git change discovery

Author: ByteVoyageur

ingestion/git_changes.py

@@ -0,0 +1,69 @@
+"""Find repository files changed since the last successful ingestion.
+
+This module reads git history as a recovery cursor: it narrows the scan to
+files touched since ``last_ingestion`` and reports old paths that disappeared
+so their vector-store chunks can be deleted. It does not decide whether a
+current file is truly changed; the pipeline still uses ``file_sha1`` for that
+state-based decision.
+"""
+
+import subprocess
+from pathlib import Path
+
+
+# Build repo -> repo-relative current candidates and stale paths from git
+# history. None candidates means "first run, full scan"; stale paths are files
+# deleted or renamed away since the last successful ingestion.
+def candidate_file_changes(repo_dirs, last_ingestion):
+    if not last_ingestion:
+        return None, {}
+    candidates = {}
+    stale = {}
+    for repo_dir in repo_dirs:
+        changed, removed = _file_changes_since(repo_dir, last_ingestion)
+        if changed:
+            candidates[Path(repo_dir).name] = changed
+        if removed:
+            stale[Path(repo_dir).name] = removed
+    return candidates, stale
+
+
+# Repo-relative file paths touched since `since`, split into current paths to
+# re-check and stale paths to delete from the vector store.
+def _file_changes_since(repo_dir, since):
+    try:
+        result = subprocess.run(
+            ["git", "-C", str(repo_dir), "log", f"--since={since}",
+             "--name-status", "--pretty=format:"],
+            capture_output=True, text=True, check=False, timeout=15)
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return set(), set()
+    if result.returncode != 0:
+        return set(), set()
+    changed = set()
+    stale = set()
+    for line in result.stdout.splitlines():
+        _collect_name_status(line, changed, stale)
+    return changed, stale
+
+
+# Interpret one git --name-status line. Add current file paths to `changed`
+# and removed/old paths to `stale`.
+def _collect_name_status(line, changed, stale):
+    parts = [part.strip().replace("\\", "/")
+             for part in line.split("\t") if part.strip()]
+    if not parts:
+        return
+    if len(parts) == 1:
+        changed.add(parts[0])
+        return
+    status = parts[0]
+    code = status[0]
+    if code == "D" and len(parts) >= 2:
+        stale.add(parts[1])
+    elif code in {"R", "C"} and len(parts) >= 3:
+        if code == "R":
+            stale.add(parts[1])
+        changed.add(parts[2])
+    elif len(parts) >= 2:
+        changed.add(parts[1])

ingestion/main.py

@@ -24,6 +24,7 @@
 from pathlib import Path
 
 from ingestion.config import IngestionConfig
+from ingestion import git_changes
 
 
 # Directories under the workspace root that are not part of the indexed
@@ -106,12 +107,14 @@ def main(argv=None):
 
     repo_dirs = _ensure_repos(root, config, argv)
     last_ingestion = vectordb.get_last_ingestion()
-    candidate_files = _candidate_files(repo_dirs, last_ingestion)
+    candidate_files, stale_files = git_changes.candidate_file_changes(
+        repo_dirs, last_ingestion)
 
     print(f"indexing {len(repo_dirs)} repo(s): "
           f"{', '.join(d.name for d in repo_dirs)}")
     if last_ingestion:
         print(f"last successful ingestion: {last_ingestion}")
+    _delete_stale_files(stale_files, vectordb)
     count = run(repo_dirs, candidate_files=candidate_files)
     vectordb.set_last_ingestion(_utc_now_iso8601())
     print(f"indexed {count} chunks into '{config.qdrant_collection}'")
@@ -129,9 +132,6 @@ def main(argv=None):
 def _ensure_repos(root, config, names=None):
     root.mkdir(parents=True, exist_ok=True)
     _ensure_development(root, config.development_repo_url)
-    repo_dirs = _discover_repos(root, names)
-    if repo_dirs:
-        return repo_dirs
     _run_k_clone(root, config.kli_organization, config.kli_workspace)
     repo_dirs = _discover_repos(root, names)
     if not repo_dirs:
@@ -190,23 +190,19 @@ def _candidate_files(repo_dirs, last_ingestion):
     return candidates
 
 
+# Delete vector-store chunks for files that disappeared from git history
+# (deleted files, or the old side of renames).
+def _delete_stale_files(stale_files, vectordb):
+    for repository, source_paths in stale_files.items():
+        for source_path in source_paths:
+            vectordb.delete_file(repository, source_path)
+
+
 # Repo-relative file paths touched since `since` (ISO 8601), newest activity
 # first in git history but returned here as a deduplicated set.
 def _git_changed_files_since(repo_dir, since):
-    try:
-        result = subprocess.run(
-            ["git", "-C", str(repo_dir), "log", f"--since={since}",
-             "--name-only", "--pretty=format:"],
-            capture_output=True, text=True, check=False, timeout=15)
-    except (FileNotFoundError, subprocess.TimeoutExpired):
-        return set()
-    if result.returncode != 0:
-        return set()
-    return {
-        line.strip().replace("\\", "/")
-        for line in result.stdout.splitlines()
-        if line.strip()
-    }
+    candidates, _ = git_changes.candidate_file_changes([repo_dir], since)
+    return candidates.get(Path(repo_dir).name, set())
 
 
 # Current UTC timestamp in compact ISO 8601 form used by the metadata