"""The Boardroom Map job runner — convenes the review panel over dropped documents.

Runs as a background thread inside the FastAPI app. It does NOT run on a clock
like Nightshift; it reacts to triggers:

  * an explicit "Run Review" (drops /data/state/run_request), or
  * autoRunOnDrop: files landing in /data/inbox, once the inbox is stable.

One job at a time. A job:
  1. extract text from the inbox locally (CPU) — only text crosses to the Sparks
  2. rsync the text to a per-job dir on the head Spark
  3. serve the needed models in WAVES; run the reviewers for each wave
  4. optionally run the local lead-reviewer synthesis
  5. pull the reports back to /data/reports/<job>, assemble latest.md
  6. wipe the documents from the Sparks (unless disabled) and tear serving down

All state (phase, current job, per-reviewer status, last report) is mirrored to
/data/state/runtime.json so the Web UI can render it.
"""
from __future__ import annotations

import json
import os
import shutil
import threading
import time
import traceback
from collections import deque
from datetime import datetime

import bm_config
import extraction
import preflight
import reviewers as rev_mod
import serving
import spark_client as sc
import synthesis as synth_mod

DATA_DIR = os.environ.get("BM_DATA_DIR", "/data")
INBOX = os.path.join(DATA_DIR, "inbox")
PROCESSED = os.path.join(DATA_DIR, "processed")
STATE_DIR = os.path.join(DATA_DIR, "state")
JOBS_DIR = os.path.join(STATE_DIR, "jobs")
REPORTS_DIR = os.path.join(DATA_DIR, "reports")
RUNTIME_PATH = os.path.join(STATE_DIR, "runtime.json")
REQUEST_PATH = os.path.join(STATE_DIR, "run_request")

TICK_SECONDS = 10


def _inbox_signature() -> tuple[int, str]:
    """(count, signature) of supported files in the inbox, for stability checks."""
    if not os.path.isdir(INBOX):
        return (0, "")
    items = []
    for fn in sorted(os.listdir(INBOX)):
        p = os.path.join(INBOX, fn)
        if os.path.isfile(p) and os.path.splitext(fn)[1].lower() in extraction.SUPPORTED:
            items.append(f"{fn}:{os.path.getsize(p)}:{int(os.path.getmtime(p))}")
    return (len(items), "|".join(items))


class JobRunner:
    def __init__(self):
        self._events = deque(maxlen=500)
        self._lock = threading.Lock()
        self.phase = "idle"   # idle | extracting | reviewing | synthesizing | collecting | done | error
        self.job_id = None
        self.message = ""
        self.panel: list[dict] = []
        self.waves_total = 0
        self.wave_index = 0
        self.last_report_path = None
        self._thread = None
        self._last_sig = None
        self._stable_sig = None
        self._last_done_sig = None
        for d in (STATE_DIR, JOBS_DIR, REPORTS_DIR, INBOX, PROCESSED):
            os.makedirs(d, exist_ok=True)
        self._restore()

    # ------------------------------------------------------------- logging
    def log(self, msg: str):
        line = f"[{datetime.now().strftime('%H:%M:%S')}] {msg}"
        with self._lock:
            self._events.append(line)
        print(line, flush=True)
        self._persist()

    def events(self) -> list[str]:
        with self._lock:
            return list(self._events)

    # ------------------------------------------------------------- persistence
    def _persist(self):
        try:
            with open(RUNTIME_PATH, "w") as f:
                json.dump(self.snapshot() | {"events": list(self._events)[-200:],
                                             "updated": time.time()}, f)
        except Exception:
            pass

    def _restore(self):
        try:
            with open(RUNTIME_PATH) as f:
                d = json.load(f)
            self.phase = d.get("phase", "idle")
            self.job_id = d.get("job_id")
            self.message = d.get("message", "")
            self.panel = d.get("panel", [])
            self.last_report_path = d.get("last_report_path")
            for e in d.get("events", []):
                self._events.append(e)
            # A job can't survive a restart; reset a stuck running phase.
            if self.phase in ("extracting", "reviewing", "synthesizing", "collecting"):
                self.phase = "idle"
        except Exception:
            pass

    def snapshot(self) -> dict:
        return {
            "phase": self.phase,
            "job_id": self.job_id,
            "message": self.message,
            "panel": self.panel,
            "waves_total": self.waves_total,
            "wave_index": self.wave_index,
            "last_report_path": self.last_report_path,
        }

    # ------------------------------------------------------------- lifecycle
    def start(self):
        if self._thread and self._thread.is_alive():
            return
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def request_run(self):
        """Public hook (used by the API) to request a review immediately."""
        try:
            with open(REQUEST_PATH, "w") as f:
                f.write(str(time.time()))
        except Exception:
            pass

    def _run(self):
        while True:
            try:
                self._poll_once()
            except Exception as e:
                self.phase = "error"
                self.message = str(e)[:300]
                self.log(f"[runner] ERROR: {e}")
                self.log(traceback.format_exc().splitlines()[-1])
            time.sleep(TICK_SECONDS)

    def _poll_once(self):
        cfg = bm_config.load()
        triggered = False
        if os.path.exists(REQUEST_PATH):
            os.remove(REQUEST_PATH)
            triggered = True
            self.log("[runner] review requested")
        elif cfg.get("autoRunOnDrop"):
            count, sig = _inbox_signature()
            if count and sig == self._last_sig and sig != self._last_done_sig:
                # stable across two ticks and not the batch we last processed
                triggered = True
                self.log("[runner] inbox stable — auto-running review")
            self._last_sig = sig

        if not triggered:
            return
        count, _ = _inbox_signature()
        if not count:
            self.log("[runner] nothing to review (inbox empty of supported files)")
            self.phase = "idle"
            self._persist()
            return
        self._run_job(cfg)

    # ------------------------------------------------------------- the job
    def _run_job(self, cfg: dict):
        job_id = datetime.now().strftime("%Y-%m-%d_%H%M%S")
        self.job_id = job_id
        self.message = ""
        self.waves_total = 0
        self.wave_index = 0
        self.panel = []
        local_job = os.path.join(JOBS_DIR, job_id)
        local_docs = os.path.join(local_job, "docs")
        remote_job = f"{cfg['remoteWorkDir'].rstrip('/')}/jobs/{job_id}"
        rubric = cfg.get("reviewInstructions") or bm_config.DEFAULT_RUBRIC
        self.log(f"=== Review job {job_id} begins ===")

        try:
            # 1. Extract locally (only text crosses to the Sparks).
            self.phase = "extracting"; self._persist()
            manifest = extraction.extract_inbox(INBOX, local_docs, self.log)
            ok_docs = [m for m in manifest if m["ok"]]
            if not ok_docs:
                raise RuntimeError("no documents could be extracted (unsupported or empty inbox)")
            self.log(f"[runner] extracted {len(ok_docs)} document(s)")

            # 2. Resolve the panel against the model catalog.
            catalog = {m["alias"] for m in (cfg.get("models") or [])}
            panel = rev_mod.roster(cfg)
            valid = [r for r in panel if r["model"] in catalog]
            invalid = [r for r in panel if r["model"] not in catalog]
            for r in invalid:
                self.log(f"[runner] WARNING: reviewer '{r['name']}' uses unknown model '{r['model']}' — skipped")
            if not valid:
                raise RuntimeError("no reviewers reference a configured model (see Configure Models/Reviewers)")
            self.panel = [{"name": r["name"], "model": r["model"], "status": "pending"} for r in valid]

            needed = {r["model"] for r in valid}
            if cfg.get("synthesisEnabled"):
                sm = synth_mod.pick_model(cfg)
                if sm:
                    needed.add(sm)

            # Air-gapped mode can't route to second-Spark models (internal net).
            if cfg.get("networkMode") == "airgapped":
                cat = {m["alias"]: m for m in (cfg.get("models") or [])}
                offenders = [a for a in needed if cat.get(a, {}).get("spark") == "secondary"]
                if offenders:
                    raise RuntimeError(
                        "air-gapped mode requires all models on the head Spark, but these are "
                        f"on the secondary: {', '.join(sorted(offenders))}. Move them to the "
                        "primary Spark or switch to local-services mode.")

            # 3. Ship text to the Spark + ensure infra.
            self.phase = "reviewing"; self._persist()
            push = sc.push_dir(sc.head(cfg), local_docs, f"{remote_job}/docs")
            if push.returncode != 0:
                raise RuntimeError(f"shipping documents to the Spark failed: {push.stderr}")
            rev_mod.ensure_reviewer_image(cfg, self.log)
            serving.ensure_network(cfg, self.log)
            # In local_services mode, warn early if the optional web_search backend
            # (SearXNG, self-signed HTTPS) is unreachable — non-fatal.
            preflight.check_searxng(cfg, self.log)

            # 4. Run the panel in waves (synthesis is handled separately, below).
            review_aliases = {r["model"] for r in valid}
            waves = serving.plan_waves(cfg, review_aliases)
            self.waves_total = len(waves)
            hf = bm_config.hf_token()
            collected = []
            for i, wave in enumerate(waves, 1):
                self.wave_index = i
                wave_aliases = {m["alias"] for m in wave}
                wpanel = [r for r in valid if r["model"] in wave_aliases]
                self.log(f"[runner] wave {i}/{len(waves)}: models={sorted(wave_aliases)} "
                         f"reviewers={[r['name'] for r in wpanel]}")
                serving.bring_up_wave(cfg, wave, hf, self.log)
                self._await_serving(cfg, wave)
                preflight.check_wave(cfg, wave, self.log)
                res = rev_mod.run_wave_reviewers(cfg, remote_job, wpanel, rubric, self.log)
                collected.extend(res)
                self._mark_panel(res)
                serving.tear_down_wave(cfg, wave, self.log)

            # 5. Synthesis (its own single-model wave).
            synth_ok = False
            if cfg.get("synthesisEnabled"):
                self.phase = "synthesizing"; self._persist()
                sm = synth_mod.pick_model(cfg)
                swave = serving.plan_waves(cfg, {sm})
                for wave in swave:
                    serving.bring_up_wave(cfg, wave, hf, self.log)
                    self._await_serving(cfg, wave)
                    preflight.check_wave(cfg, wave, self.log)
                    sres = synth_mod.run_synthesis(cfg, remote_job, rubric, self.log)
                    synth_ok = bool(sres.get("report"))
                    serving.tear_down_wave(cfg, wave, self.log)

            # 6. Collect reports + assemble.
            self.phase = "collecting"; self._persist()
            self._collect(cfg, job_id, remote_job, local_job, valid, manifest, synth_ok)

            # 7. Confidentiality: wipe the documents from the Spark.
            if cfg.get("wipeRemoteDocs", True):
                sc.run(sc.head(cfg), f"rm -rf {remote_job}", timeout=60)
                self.log("[runner] wiped document text from the Spark")
            serving.tear_down_all(cfg, self.log)

            # 8. Clear the inbox (move originals aside so they aren't re-reviewed).
            self._drain_inbox(job_id)
            self._last_done_sig = _inbox_signature()[1]
            self.phase = "done"
            self.message = f"Reviewed {len(ok_docs)} document(s) with {len(valid)} reviewer(s)."
            self.log(f"=== Review job {job_id} complete ===")
            self._persist()
        except Exception as e:
            self.phase = "error"
            self.message = f"Review failed: {e}"
            self.log(f"[runner] JOB FAILED — {e}")
            self.log(traceback.format_exc().splitlines()[-1])
            try:
                serving.tear_down_all(cfg, self.log)
            except Exception:
                pass
            self._persist()

    # ------------------------------------------------------------- helpers
    def _await_serving(self, cfg, wave, timeout=900):
        self.log("[runner] waiting for wave serving to come online…")
        deadline = time.time() + timeout
        want = len(wave) + 1  # vLLMs + proxy
        while time.time() < deadline:
            running = serving.health(cfg).get("running", [])
            if sum(1 for r in running if "Up" in r) >= want:
                self.log("[runner] wave serving online")
                return
            time.sleep(15)
        self.log("[runner] WARNING: wave serving not fully confirmed; continuing")

    def _mark_panel(self, res: list[dict]):
        by_name = {r["name"]: r for r in res}
        for p in self.panel:
            r = by_name.get(p["name"])
            if not r:
                continue
            if not r.get("ok", True):
                p["status"] = "launch-failed"
            elif r.get("report"):
                p["status"] = "done"
            else:
                p["status"] = "no-report"
        self._persist()

    def _collect(self, cfg, job_id, remote_job, local_job, valid, manifest, synth_ok):
        out_local = os.path.join(REPORTS_DIR, job_id)
        os.makedirs(out_local, exist_ok=True)
        sc.pull_dir(sc.head(cfg), f"{remote_job}/out", os.path.join(out_local, "reviewers"))
        if synth_ok:
            sc.pull_dir(sc.head(cfg), f"{remote_job}/synth-out", os.path.join(out_local, "synthesis"))

        # Assemble a single latest.md: the consolidated report if present, else a
        # concatenation of the individual reports.
        parts = [f"# Boardroom Map review — {job_id}\n",
                 "Documents reviewed: " + ", ".join(m["source"] for m in manifest if m["ok"]) + "\n",
                 "Panel: " + ", ".join(f"{r['name']} ({r['model']})" for r in valid) + "\n"]
        consolidated = os.path.join(out_local, "synthesis", "CONSOLIDATED_REPORT.md")
        if synth_ok and os.path.exists(consolidated):
            parts.append("\n---\n\n## Consolidated report (lead reviewer)\n\n")
            parts.append(open(consolidated, errors="replace").read())
            parts.append("\n\n---\n")
        parts.append("\n## Individual reviewer reports\n")
        rev_local = os.path.join(out_local, "reviewers")
        if os.path.isdir(rev_local):
            for fn in sorted(os.listdir(rev_local)):
                if fn.endswith(".md"):
                    parts.append(f"\n### {fn[:-3]}\n\n")
                    parts.append(open(os.path.join(rev_local, fn), errors="replace").read())
                    parts.append("\n")
        assembled = "".join(parts)
        with open(os.path.join(out_local, "report.md"), "w") as f:
            f.write(assembled)
        with open(os.path.join(REPORTS_DIR, "latest.md"), "w") as f:
            f.write(assembled)
        self.last_report_path = os.path.join(out_local, "report.md")
        self.log(f"[runner] reports saved to {out_local}")

    def _drain_inbox(self, job_id):
        dest = os.path.join(PROCESSED, job_id)
        os.makedirs(dest, exist_ok=True)
        for fn in os.listdir(INBOX):
            src = os.path.join(INBOX, fn)
            if os.path.isfile(src):
                try:
                    shutil.move(src, os.path.join(dest, fn))
                except Exception:
                    pass
        self.log(f"[runner] inbox cleared (originals moved to processed/{job_id})")


# Module-level singleton used by app.py
runner = JobRunner()