Update main.py

main.py

@@ -7,6 +7,9 @@ import uuid
 import random
 import logging
 import hashlib
+import json
+import tempfile
+import threading
 from typing import List, Optional, Dict, Any, Tuple
 
 import numpy as np
@@ -69,10 +72,15 @@ WIPE_BEFORE_INDEX = os.getenv("WIPE_BEFORE_INDEX", "false").lower() in ("1","tru
 # --- Auth d’API de ce service (simple header) ---
 AUTH_TOKEN = os.getenv("REMOTE_INDEX_TOKEN", "").strip()
 
+# --- Jobs persistence / runtime behavior ---
+JOBS_FILE = os.getenv("JOBS_FILE", "/tmp/remote_index_jobs.json")
+RUN_INDEX_BLOCKING = os.getenv("RUN_INDEX_BLOCKING", "false").lower() in ("1","true","yes","on")
+
 LOG.info(f"Embeddings backend order = {EMB_BACKEND_ORDER}")
 LOG.info(f"HF pipeline URL = {HF_URL_PIPE}")
 LOG.info(f"HF models   URL = {HF_URL_MODL}")
 LOG.info(f"VECTOR_STORE = {VECTOR_STORE}")
+LOG.info(f"JOBS_FILE = {JOBS_FILE} RUN_INDEX_BLOCKING = {RUN_INDEX_BLOCKING}")
 
 if "deepinfra" in EMB_BACKEND_ORDER and not DI_TOKEN:
     LOG.warning("DEEPINFRA_API_KEY manquant — tentatives DeepInfra échoueront.")
@@ -293,19 +301,59 @@ class StatusBody(BaseModel):
     job_id: str
 
 # ======================================================================================
-# Jobs store (mémoire)
+# Jobs store (persistence simple)
 # ======================================================================================
 JOBS: Dict[str, Dict[str, Any]] = {}  # {job_id: {"status": "...", "logs": [...], "created": ts}}
+_jobs_lock = threading.Lock()
+
+def _atomic_write(path: str, data: str):
+    dirn = os.path.dirname(path) or "/tmp"
+    fd, tmp_path = tempfile.mkstemp(dir=dirn)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(data)
+        os.replace(tmp_path, path)
+    finally:
+        if os.path.exists(tmp_path):
+            try:
+                os.unlink(tmp_path)
+            except Exception:
+                pass
+
+def _load_jobs():
+    global JOBS
+    try:
+        if os.path.exists(JOBS_FILE):
+            with open(JOBS_FILE, "r", encoding="utf-8") as f:
+                JOBS = json.load(f)
+            LOG.info("Jobs loaded from %s (%d entries)", JOBS_FILE, len(JOBS))
+    except Exception as e:
+        LOG.warning("Could not load jobs file %s: %s", JOBS_FILE, e)
+
+def _save_jobs():
+    try:
+        with _jobs_lock:
+            _atomic_write(JOBS_FILE, json.dumps(JOBS, ensure_ascii=False))
+    except Exception as e:
+        LOG.warning("Could not save jobs file %s: %s", JOBS_FILE, e)
+
+# load existing jobs at startup
+_load_jobs()
 
 def _append_log(job_id: str, line: str):
     job = JOBS.get(job_id)
-    if job:
-        job["logs"].append(line)
+    if job is not None:
+        job.setdefault("logs", []).append(line)
+        # keep logs bounded to avoid huge files
+        if len(job["logs"]) > 5000:
+            job["logs"] = job["logs"][-5000:]
+        _save_jobs()
 
 def _set_status(job_id: str, status: str):
     job = JOBS.get(job_id)
-    if job:
+    if job is not None:
         job["status"] = status
+        _save_jobs()
 
 def _auth(x_auth: Optional[str]):
     if AUTH_TOKEN and (x_auth or "") != AUTH_TOKEN:
@@ -484,7 +532,7 @@ def _chunk_with_spans(text: str, size: int, overlap: int):
             break
 
 # ======================================================================================
-# Background task : indexation — VERSION CORRIGÉE
+# Background task : indexation — VERSION CORRIGÉE (persistance JOBS)
 # ======================================================================================
 
 def run_index_job(job_id: str, req: IndexRequest):
@@ -706,9 +754,16 @@ def start_index(req: IndexRequest, background_tasks: BackgroundTasks, x_auth_tok
     _check_backend_ready()
     job_id = uuid.uuid4().hex[:12]
     JOBS[job_id] = {"status": "queued", "logs": [], "created": time.time()}
+    _save_jobs()
     LOG.info(f"Created job {job_id} for project {req.project_id}")
     _append_log(job_id, f"Job created: {job_id} project={req.project_id}")
-    background_tasks.add_task(run_index_job, job_id, req)
+
+    if RUN_INDEX_BLOCKING:
+        # utile pour environnements où les BackgroundTasks ne persistent pas / sont killés
+        run_index_job(job_id, req)
+    else:
+        background_tasks.add_task(run_index_job, job_id, req)
+
     return {"job_id": job_id}
 
 # --- 3 variantes pour /status ---
@@ -719,7 +774,7 @@ def status_path(job_id: str, x_auth_token: Optional[str] = Header(default=None))
     if not j:
         # Response JSON plus explicite pour faciliter le debug côté client
         raise HTTPException(status_code=404, detail={"error": "job inconnu", "advice": "POST /index to create a new job"})
-    return {"status": j["status"], "logs": j["logs"][-1500:]}
+    return {"status": j["status"], "logs": j.get("logs", [])[-1500:]}
 
 @app.get("/status")
 def status_query(job_id: str = Query(...), x_auth_token: Optional[str] = Header(default=None)):