fix

ragg/ingest_trigger.py

@@ -23,6 +23,21 @@ def _list_dir_sample(path: str, patterns=("*.pdf", "*.PDF"), limit: int = 10):
     except Exception as e:
         return {"exists": False, "error": str(e)}
 
+def _count_files(path: str, exts=(".json", ".sqlite3", ".parquet", ".bin", ".txt", ".lock")):
+    try:
+        if not os.path.isdir(path):
+            return {"exists": False, "total": 0, "by_ext": {}}
+        counts = {}
+        total = 0
+        for root, _, files in os.walk(path):
+            for f in files:
+                total += 1
+                ext = os.path.splitext(f)[1].lower()
+                counts[ext] = counts.get(ext, 0) + 1
+        return {"exists": True, "total": total, "by_ext": counts}
+    except Exception as e:
+        return {"exists": False, "total": 0, "error": str(e)}
+
 @ingest_trigger_bp.route("/ingest/ping", methods=["GET"])
 def ingest_ping():
     info = {
@@ -41,7 +56,6 @@ def ingest_ping():
     print(info, flush=True)
     return jsonify(info), 200
 
-# ---------- LIGHTWEIGHT DEBUG (no ingestion run) ----------
 @ingest_trigger_bp.route("/ingest/debug", methods=["GET"])
 def ingest_debug():
     import importlib
@@ -78,7 +92,18 @@ def ingest_debug():
     print(resp, flush=True)
     return jsonify(resp), 200
 
-# ---------- INLINE RUN (use only for debugging) ----------
+@ingest_trigger_bp.route("/ingest/status", methods=["GET"])
+def ingest_status():
+    chroma_root = os.getenv("CHROMA_ROOT") or "/data/chroma"
+    levels = ["low", "mid", "high"]
+    out = {"chroma_root": chroma_root, "levels": {}}
+    for lv in levels:
+        out["levels"][lv] = _count_files(os.path.join(chroma_root, lv))
+    print("\n=== [INGEST STATUS] ===", flush=True)
+    print(out, flush=True)
+    return jsonify(out), 200
+
+# Inline run (explicit)
 @ingest_trigger_bp.route("/ingest/run-inline", methods=["POST"])
 def ingest_run_inline():
     import importlib
@@ -90,8 +115,8 @@ def ingest_run_inline():
         if not hasattr(mod, "ingest_all_levels"):
             return jsonify({"status": "error", "message": "ingest_all_levels() not found"}), 500
 
-        # Optional: quick preflight
-        for p in ["ragg/pdfs/low", "ragg/pdfs/mid", "ragg/pdfs/high"]:
+        # Quick preflight
+        for p in ["ragg/pdfs/low", "ragg/pdfs/mid", "ragg/pdfs/high", "pdfs/low", "pdfs/mid", "pdfs/high"]:
             print(f"[INLINE] {p} -> {_list_dir_sample(p)}", flush=True)
 
         mod.ingest_all_levels()
@@ -102,6 +127,7 @@ def ingest_run_inline():
         return jsonify({"status": "error", "error": f"{type(e).__name__}: {e}",
                         "traceback": traceback.format_exc()}), 500
 
+# /ingest – inline by default; subprocess kept as optional fallback
 @ingest_trigger_bp.route("/ingest", methods=["POST"])
 def run_ingest():
     if not _ingest_lock.acquire(blocking=False):
@@ -109,40 +135,40 @@ def run_ingest():
         return jsonify({"status": "busy", "message": "Ingestion already in progress"}), 409
 
     try:
+        use_subprocess = os.getenv("INGEST_USE_SUBPROCESS", "0") == "1"
         module_name = os.getenv("INGEST_MODULE", "ragg.ingest_all")
         timeout_sec = int(os.getenv("INGEST_TIMEOUT_SEC", "1800"))
 
         print("\n=== [INGEST TRIGGER] ===", flush=True)
         print("Trigger called", flush=True)
+        print("Use subprocess:", use_subprocess, flush=True)
         print("Module:", module_name, flush=True)
         print("CWD:", os.getcwd(), flush=True)
         print("PYTHON:", sys.executable, flush=True)
         print("PYTHONPATH (len={}):".format(len(sys.path)), flush=True)
         print(sys.path, flush=True)
-
-        # Log key env vars the ingestion relies on
         print("ENV:", os.getenv("ENV"), flush=True)
         print("CHROMA_DIR:", os.getenv("CHROMA_DIR"), flush=True)
         print("CHROMA_ROOT:", os.getenv("CHROMA_ROOT"), flush=True)
-        print("HF_HOME:", os.getenv("HF_HOME"), flush=True)
-
-        # Quick preflight: show whether expected PDF folders exist and sample files
-        bases = [
-            "ragg/pdfs/low",
-            "ragg/pdfs/mid",
-            "ragg/pdfs/high",
-            "pdfs/low",
-            "pdfs/mid",
-            "pdfs/high",
-        ]
-        print("\n[DEBUG] PDF folders preflight:", flush=True)
-        for b in bases:
+
+        # Preflight: show expected PDF folders
+        for b in ["ragg/pdfs/low", "ragg/pdfs/mid", "ragg/pdfs/high", "pdfs/low", "pdfs/mid", "pdfs/high"]:
             info = _list_dir_sample(b)
             print(f"  - {b}: exists={info.get('exists')} total={info.get('total', 0)} sample={info.get('sample', [])}", flush=True)
 
+        if not use_subprocess:
+            # Inline mode (default)
+            import importlib
+            print("[DEBUG] Running inline ingestion...", flush=True)
+            mod = importlib.import_module(module_name)
+            if not hasattr(mod, "ingest_all_levels"):
+                return jsonify({"status": "error", "message": "ingest_all_levels() not found"}), 500
+            mod.ingest_all_levels()
+            return jsonify({"status": "success", "mode": "inline"}), 200
+
+        # Subprocess fallback (set INGEST_USE_SUBPROCESS=1 to enable)
         cmd = [sys.executable, "-m", module_name]
         print("\n[DEBUG] Running subprocess:", cmd, flush=True)
-
         result = subprocess.run(
             cmd,
             capture_output=True,
@@ -150,7 +176,6 @@ def run_ingest():
             cwd=None,
             timeout=timeout_sec
         )
-
         print("\n[DEBUG] Subprocess completed.", flush=True)
         print("Return code:", result.returncode, flush=True)
 
@@ -173,6 +198,7 @@ def run_ingest():
             "returncode": result.returncode,
             "stdout": (result.stdout or "")[-4000:],
             "stderr": (result.stderr or "")[-4000:],
+            "mode": "subprocess"
         }
         status_code = 200 if result.returncode == 0 else 500
         return jsonify(payload), status_code