Implement graph edges collection support with admin and ingest updates

scripts/collection_admin.py

@@ -193,6 +193,7 @@ def delete_collection_everywhere(
     out: Dict[str, Any] = {
         "collection": name,
         "qdrant_deleted": False,
+        "qdrant_graph_deleted": False,
         "registry_marked_deleted": False,
         "deleted_state_files": 0,
         "deleted_managed_workspaces": 0,
@@ -209,6 +210,14 @@ def delete_collection_everywhere(
                     out["qdrant_deleted"] = True
                 except Exception:
                     out["qdrant_deleted"] = False
+                # Best-effort: also delete companion graph edges collection when present.
+                # This branch stores file-level edges in `<collection>_graph`.
+                if not name.endswith("_graph"):
+                    try:
+                        cli.delete_collection(collection_name=f"{name}_graph")
+                        out["qdrant_graph_deleted"] = True
+                    except Exception:
+                        out["qdrant_graph_deleted"] = False
     except Exception:
         out["qdrant_deleted"] = False
 
@@ -359,8 +368,10 @@ def _manual_copy_points() -> None:
                 vectors_config = None
                 sparse_vectors_config = None
 
+            # Support vector-less collections (e.g. payload-only graph edge collections).
             if vectors_config is None:
-                raise RuntimeError(f"Cannot determine vectors config for source collection {src}")
+                vectors_config = {}
+            vectorless = isinstance(vectors_config, dict) and not vectors_config
 
             try:
                 cli.create_collection(
@@ -401,7 +412,7 @@ def _manual_copy_points() -> None:
                         limit=batch_limit,
                         offset=offset,
                         with_payload=True,
-                        with_vectors=True,
+                        with_vectors=(not vectorless),
                     )
                 except Exception as exc:
                     raise RuntimeError(f"Failed to scroll points from {src}: {exc}") from exc
@@ -414,7 +425,9 @@ def _manual_copy_points() -> None:
                         point_id = getattr(record, "id", None)
                         payload = getattr(record, "payload", None)
                         vector = None
-                        if hasattr(record, "vector") and getattr(record, "vector") is not None:
+                        if vectorless:
+                            vector = {}
+                        elif hasattr(record, "vector") and getattr(record, "vector") is not None:
                             vector = getattr(record, "vector")
                         elif hasattr(record, "vectors") and getattr(record, "vectors") is not None:
                             vector = getattr(record, "vectors")
@@ -477,4 +490,23 @@ def _count_points(name: str) -> Optional[int]:
         # The manual path guarantees the destination gets the exact same points/payloads/vectors.
         _manual_copy_points()
 
+    # Best-effort: copy the companion graph collection when copying a base collection.
+    # Graph edges are derived data and can be rebuilt, but copying avoids a cold-start window
+    # during staging cutovers where the clone has no graph.
+    if not src.endswith("_graph") and not dest.endswith("_graph"):
+        try:
+            copy_collection_qdrant(
+                source=f"{src}_graph",
+                target=f"{dest}_graph",
+                qdrant_url=base_url,
+                overwrite=overwrite,
+            )
+        except Exception as exc:
+            logger.debug(
+                "Best-effort graph collection copy %s_graph -> %s_graph failed: %s",
+                src,
+                dest,
+                exc,
+            )
+
     return dest

scripts/indexing_admin.py

@@ -927,6 +927,17 @@ def delete_collection_qdrant(*, qdrant_url: str, api_key: Optional[str], collect
         return
     try:
         cli.delete_collection(collection_name=name)
+        # Best-effort: also delete companion graph edges collection when present.
+        if not name.endswith("_graph"):
+            try:
+                cli.delete_collection(collection_name=f"{name}_graph")
+            except Exception as exc:
+                try:
+                    print(
+                        f"[indexing_admin] best-effort graph collection delete failed for {name}_graph: {exc}"
+                    )
+                except Exception:
+                    pass
     except Exception:
         pass
     finally:
@@ -951,6 +962,17 @@ def recreate_collection_qdrant(*, qdrant_url: str, api_key: Optional[str], colle
             cli.delete_collection(collection_name=name)
         except Exception as delete_error:
             raise RuntimeError(f"Failed to delete existing collection '{name}' in Qdrant: {delete_error}") from delete_error
+        # Best-effort: also delete companion graph edges collection when present.
+        if not name.endswith("_graph"):
+            try:
+                cli.delete_collection(collection_name=f"{name}_graph")
+            except Exception as exc:
+                try:
+                    print(
+                        f"[indexing_admin] best-effort graph collection delete failed for {name}_graph: {exc}"
+                    )
+                except Exception:
+                    pass
     finally:
         try:
             cli.close()
@@ -984,12 +1006,9 @@ def spawn_ingest_code(
                 env.pop(k, None)
             else:
                 env[str(k)] = str(v)
-        # When we provide env overrides for a run (e.g. staging rebuild), we also want to
-        # force ingest_code to honor the explicit COLLECTION_NAME instead of routing based
-        # on per-repo state/serving_collection in multi-repo mode.
-        # CTXCE_FORCE_COLLECTION_NAME is only used for these subprocess runs; normal watcher
-        # and indexer flows do not set it.
-        env["CTXCE_FORCE_COLLECTION_NAME"] = "1"  # Force ingest_code to use COLLECTION_NAME for staging/pending env overrides
+    # For admin-triggered subprocess runs (recreate/reindex/staging), force ingest_code to
+    # honor explicit COLLECTION_NAME and avoid multi-repo enumeration.
+    env["CTXCE_FORCE_COLLECTION_NAME"] = "1"
     env["COLLECTION_NAME"] = collection
     env["WATCH_ROOT"] = work_dir
     env["WORKSPACE_PATH"] = work_dir

scripts/ingest/cli.py

@@ -15,6 +15,8 @@
     is_multi_repo_mode,
     get_collection_name,
 )
+from scripts import workspace_state as _ws
+from scripts.collection_health import clear_indexing_caches as _clear_indexing_caches_impl
 from scripts.ingest.pipeline import index_repo
 from scripts.ingest.pseudo import generate_pseudo_tags
 
@@ -40,6 +42,11 @@ def parse_args():
         action="store_true",
         help="Do not skip files whose content hash matches existing index",
     )
+    parser.add_argument(
+        "--clear-indexing-caches",
+        action="store_true",
+        help="Clear local indexing caches (file hash/symbol caches) before indexing",
+    )
     parser.add_argument(
         "--schema-mode",
         type=str,
@@ -186,13 +193,25 @@ def main():
         )
         return
 
+    def _clear_indexing_caches(workspace_root: Path, repo_name: str | None) -> None:
+        try:
+            _clear_indexing_caches_impl(str(workspace_root), repo_name=repo_name)
+        except Exception:
+            pass
+
     qdrant_url = os.environ.get("QDRANT_URL", "http://localhost:6333")
     api_key = os.environ.get("QDRANT_API_KEY")
     collection = os.environ.get("COLLECTION_NAME") or os.environ.get("DEFAULT_COLLECTION") or "codebase"
     model_name = os.environ.get("EMBEDDING_MODEL", "BAAI/bge-base-en-v1.5")
 
     # Resolve collection name based on multi-repo mode
-    multi_repo = bool(is_multi_repo_mode and is_multi_repo_mode())
+    force_collection = (os.environ.get("CTXCE_FORCE_COLLECTION_NAME") or "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "on",
+    }
+    multi_repo = bool(is_multi_repo_mode and is_multi_repo_mode()) and not force_collection
     if multi_repo:
         print("[multi_repo] Multi-repo mode enabled - will create separate collections per repository")
 
@@ -231,6 +250,9 @@ def main():
             if not repo_collection:
                 repo_collection = "codebase"
 
+            if args.clear_indexing_caches:
+                _clear_indexing_caches(root_path, repo_name)
+
             index_repo(
                 repo_root,
                 qdrant_url,
@@ -249,7 +271,7 @@ def main():
             try:
                 resolved = get_collection_name(str(Path(args.root).resolve()))
                 placeholders = {"", "default-collection", "my-collection", "codebase"}
-                if resolved and collection in placeholders:
+                if resolved and collection in placeholders and not force_collection:
                     collection = resolved
             except Exception:
                 pass
@@ -260,6 +282,9 @@ def main():
     flag = (os.environ.get("PSEUDO_DEFER_TO_WORKER") or "").strip().lower()
     pseudo_mode = "off" if flag in {"1", "true", "yes", "on"} else "full"
 
+    if args.clear_indexing_caches:
+        _clear_indexing_caches(Path(args.root).resolve(), None)
+
     index_repo(
         Path(args.root).resolve(),
         qdrant_url,