fallout-venice/dashboard/process_mgr.py

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