import os import subprocess import threading _procs = {} _procs_lock = threading.Lock() _PROC_DEFS = { "sim_enricher": { "label": "Enrichissement règles sim (LLM → sim_proposals)", "cmd": ["python3", "-u", "/opt/coyote/pipelines/sim_enricher.py", "--type", "all"], "env_extra": {"CHROMA_URL": "http://chromadb:8000"}, }, "lore_enricher": { "label": "Enrichissement lore (LLM → lore_proposals)", "cmd": ["python3", "-u", "/app/src/engine/lore_enricher.py", "--faction", "all"], "env_extra": {"CHROMA_URL": "http://chromadb:8000"}, }, "embed_fallout": { "label": "Re-embed PDFs Fallout → fallout_lore", "cmd": ["python3", "-u", "/opt/coyote/pipelines/embed_fallout.py"], "env_extra": {"CHROMA_URL": "http://chromadb:8000"}, }, "embed_sim": { "label": "Re-embed règles sim → fallout_sim_rules", "cmd": ["python3", "-u", "/opt/coyote/pipelines/embed_sim.py"], "env_extra": {"CHROMA_URL": "http://chromadb:8000"}, }, } def _run_proc_loop_bg(name, cmd, env_extra=None, count=1): env = {**os.environ, **(env_extra or {}), "PYTHONUNBUFFERED": "1"} with _procs_lock: _procs[name] = {"proc": None, "output": [], "running": True, "returncode": None, "error": None, "iteration": 0, "total": count} iteration = 0 try: while count == 0 or iteration < count: iteration += 1 sep = f"─── Itération {iteration}" + (f"/{count}" if count else "/∞") + " ───" with _procs_lock: _procs[name]["output"].append(sep) _procs[name]["iteration"] = iteration proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env) with _procs_lock: _procs[name]["proc"] = proc for line in proc.stdout: with _procs_lock: _procs[name]["output"].append(line.rstrip()) proc.wait() if proc.returncode is not None and proc.returncode < 0: break with _procs_lock: if not _procs[name]["running"]: break rc = proc.returncode if rc != 0: with _procs_lock: _procs[name]["returncode"] = rc break with _procs_lock: if _procs[name]["returncode"] is None: _procs[name]["returncode"] = 0 _procs[name]["running"] = False except Exception as e: with _procs_lock: _procs[name]["error"] = str(e) _procs[name]["running"] = False def _run_mix_bg(count=10): name = "mix" sim_cmd = list(_PROC_DEFS["sim_enricher"]["cmd"]) lore_cmd = list(_PROC_DEFS["lore_enricher"]["cmd"]) env = {**os.environ, "CHROMA_URL": "http://chromadb:8000", "PYTHONUNBUFFERED": "1"} with _procs_lock: _procs[name] = {"proc": None, "output": [], "running": True, "returncode": None, "error": None, "iteration": 0, "total": count * 2} iteration = 0 try: for script_name, cmd in [("sim_enricher", sim_cmd), ("lore_enricher", lore_cmd)]: i = 0 while count == 0 or i < count: i += 1 iteration += 1 sep = f"─── MIX {script_name} {i}" + (f"/{count}" if count else "/∞") + " ───" with _procs_lock: if not _procs[name]["running"]: raise StopIteration _procs[name]["output"].append(sep) _procs[name]["iteration"] = iteration proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env) with _procs_lock: _procs[name]["proc"] = proc for line in proc.stdout: with _procs_lock: _procs[name]["output"].append(line.rstrip()) proc.wait() if proc.returncode is not None and proc.returncode < 0: raise StopIteration with _procs_lock: if not _procs[name]["running"]: raise StopIteration with _procs_lock: _procs[name]["returncode"] = 0 _procs[name]["running"] = False except StopIteration: with _procs_lock: _procs[name]["returncode"] = 0 _procs[name]["running"] = False except Exception as e: with _procs_lock: _procs[name]["error"] = str(e) _procs[name]["running"] = False def proc_start(name: str, cmd: list, env_extra: dict, count: int): t = threading.Thread(target=_run_proc_loop_bg, args=(name, cmd, env_extra, count), daemon=True) t.start() def proc_mix_start(count: int): t = threading.Thread(target=_run_mix_bg, args=(count,), daemon=True) t.start() def proc_get_state(name: str) -> dict: with _procs_lock: state = dict(_procs.get(name, {"running": False, "output": [], "returncode": None, "error": None})) state.pop("proc", None) return state def proc_is_running(name: str) -> bool: with _procs_lock: return _procs.get(name, {}).get("running", False) def proc_stop(name: str): with _procs_lock: p = _procs.get(name, {}).get("proc") if name in _procs: _procs[name]["running"] = False if p: try: p.terminate() except Exception: pass