"""The Boardroom Map job runner — grades dropped board decks against the BDEF.

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

  * an explicit "Grade Decks" run request (drops /data/state/run_request), or
  * autoRunOnDrop: files landing under /data/inbox/<company-slug>/, once the
    inbox is stable across two ticks.

One job at a time. A job iterates the discovered deck units OLDEST FIRST per
company, and for each deck:

  1. extract text locally (CPU) — only text crosses to the Sparks
  2. rsync the per-deck bundle (docs/, BDEF.md, personas/, schemas/, out/,
     adjudicator-out/) to {remoteWorkDir}/jobs/<job>/<company>/<deck>/
  3. serve the needed models in WAVES (graders' models ∪ the extractor model);
     the extractor runs when its model's wave is up, graders in their waves
  4. pull out/, validate: extraction.json invalid => the DECK fails (the job
     continues); grader JSONs are validated individually, invalid ones dropped;
     fewer than 2 valid grade reports => the deck fails
  5. adjudicate (optional, non-fatal): a local model weighs the panel's evidence
  6. score deterministically (scoring.score_deck) against the company's pinned
     targets + the prior deck's forward targets, and record it in the ledger
  7. render DECK_REPORT.md + refresh the company SCORECARD.md and the
     /data/reports copies

Then it wipes the remote job dir (unless disabled), tears serving down, and
moves the graded originals to /data/processed/<job>/<slug>/ (the company folder
stays in the inbox for reuse). One deck's failure never kills the job: the job
ends "done" if at least one deck was graded.

All state (phase, per-deck status, panel 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 re
import shutil
import threading
import time
import traceback
from collections import deque
from datetime import datetime, timezone

import adjudicator as adj_mod
import bm_config
import decks
import extraction
import graders as gr_mod
import ledger as ledger_mod
import preflight
import scorecard
import scoring
import serving
import spark_client as sc
import validate

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")
LEDGER_DIR = os.path.join(DATA_DIR, "ledger")
RUNTIME_PATH = os.path.join(STATE_DIR, "runtime.json")
REQUEST_PATH = os.path.join(STATE_DIR, "run_request")

TICK_SECONDS = 10
RUNNING_PHASES = ("extracting", "grading", "adjudicating", "scoring", "collecting")

def _inbox_signature() -> tuple[int, str]:
    """(count, signature) of supported files anywhere in the inbox tree, for
    autoRunOnDrop stability checks (decks live in per-company subfolders)."""
    if not os.path.isdir(INBOX):
        return (0, "")
    items = []
    for root, _dirs, files in os.walk(INBOX):
        for fn in files:
            p = os.path.join(root, fn)
            if os.path.splitext(fn)[1].lower() in extraction.SUPPORTED:
                try:
                    items.append(f"{os.path.relpath(p, INBOX)}:{os.path.getsize(p)}:"
                                 f"{int(os.path.getmtime(p))}")
                except OSError:
                    pass
    items.sort()
    return (len(items), "|".join(items))


def _token(s: str) -> str:
    t = re.sub(r"[^a-z0-9]+", "-", (s or "").lower()).strip("-")
    return t or "deck"


def _sparks_disjoint(cfg: dict, extractor_alias: str, wpanel: list[dict]) -> bool:
    """True when the extractor's model and every grader model in the wave sit on
    different Sparks (separate GPUs) — extraction can then overlap grading."""
    catalog = {m["alias"]: m for m in (cfg.get("models") or [])}
    ex_spark = (catalog.get(extractor_alias) or {}).get("spark") or "primary"
    return all(((catalog.get(r["model"]) or {}).get("spark") or "primary") != ex_spark
               for r in wpanel)


def _composite(record) -> float | None:
    """Best-effort composite lookup on the scoring record (shape owned by scoring.py)."""
    if not isinstance(record, dict):
        return None
    for k in ("composite", "composite_score"):
        v = record.get(k)
        if isinstance(v, (int, float)):
            return v
    for parent in ("scores", "totals", "score"):
        d = record.get(parent)
        if isinstance(d, dict) and isinstance(d.get("composite"), (int, float)):
            return d["composite"]
    return None


class JobRunner:
    def __init__(self):
        self._events = deque(maxlen=500)
        self._lock = threading.Lock()
        self.phase = "idle"   # idle | extracting | grading | adjudicating | scoring | collecting | done | error
        self.job_id = None
        self.message = ""
        self.panel: list[dict] = []
        self.waves_total = 0
        self.wave_index = 0
        self.decks_total = 0
        self.deck_index = 0
        self.company = None
        self.period = None
        self.decks: list[dict] = []
        self.last_report_path = None
        self._live_aliases: set[str] | None = None  # kept-warm wave's model aliases
        self._thread = None
        self._last_sig = None
        self._last_done_sig = None
        for d in (STATE_DIR, JOBS_DIR, REPORTS_DIR, LEDGER_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.decks = d.get("decks", [])
            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 RUNNING_PHASES:
                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,
            "decks_total": self.decks_total,
            "deck_index": self.deck_index,
            "company": self.company,
            "period": self.period,
            "decks": self.decks,
            "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 grading run 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] grading run 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
                if decks.discover(INBOX).get("units"):
                    triggered = True
                    self.log("[runner] inbox stable — auto-running grading")
                else:
                    # only ungradeable files (e.g. dropped at the inbox root):
                    # mark the batch seen so we don't retrigger every tick
                    self._last_done_sig = sig
            self._last_sig = sig

        if not triggered:
            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.decks_total = 0
        self.deck_index = 0
        self.company = None
        self.period = None
        self.decks = []
        self.panel = []
        self._live_aliases = None
        remote_root = f"{cfg['remoteWorkDir'].rstrip('/')}/jobs/{job_id}"
        self.log(f"=== Grading job {job_id} begins ===")

        try:
            # 1. Discover deck units (per-company folders; oldest first).
            disc = decks.discover(INBOX)
            units = disc.get("units") or []
            for fn in disc.get("skipped") or []:
                self.log(f"[runner] WARNING: skipping root-level inbox file '{fn}' — "
                         "decks belong in /data/inbox/<company-slug>/")
            if not units:
                raise RuntimeError("nothing to grade — drop decks into /data/inbox/<company-slug>/")

            # 2. Ledger, merged with the configured portfolio companies.
            led = ledger_mod.Ledger(LEDGER_DIR)
            led.merge_config_companies(cfg.get("companies") or [])

            # Resolve the grader panel + extractor model against the catalog.
            models = cfg.get("models") or []
            catalog = {m["alias"] for m in models}
            panel = gr_mod.roster(cfg)
            valid = [r for r in panel if r["model"] in catalog]
            for r in panel:
                if r["model"] not in catalog:
                    self.log(f"[runner] WARNING: grader '{r['name']}' uses unknown model "
                             f"'{r['model']}' — skipped")
            if len(valid) < 2:
                raise RuntimeError(
                    "need at least 2 graders referencing configured models — every deck "
                    "requires >= 2 valid grade reports (see Configure Models/Graders)")
            extractor_model = (cfg.get("extractorModel") or "").strip() or \
                (models[0]["alias"] if models else "")
            if extractor_model not in catalog:
                raise RuntimeError(f"extractor model '{extractor_model}' is not in the model catalog")

            needed = {r["model"] for r in valid} | {extractor_model}
            adjudicate = bool(cfg.get("adjudicatorEnabled"))
            adj_model = adj_mod.pick_model(cfg) if adjudicate else ""
            if adjudicate and adj_model not in catalog:
                self.log(f"[runner] WARNING: adjudicator model '{adj_model}' is not in the "
                         "model catalog — adjudication disabled for this job")
                adjudicate, adj_model = False, ""
            needed_all = needed | ({adj_model} if adjudicate and adj_model else set())

            # Second-Spark models need the secondary configured. (In air-gapped
            # mode the proxy is dual-homed onto the bridge to reach them — the
            # grader containers themselves stay on the zero-egress internal net.)
            cat = {m["alias"]: m for m in models}
            on_secondary = [a for a in needed_all if cat.get(a, {}).get("spark") == "secondary"]
            if on_secondary and not (cfg.get("useBothSparks") and cfg.get("secondarySparkHost")):
                raise RuntimeError(
                    f"model(s) {', '.join(sorted(on_secondary))} are assigned to the "
                    "secondary Spark, but no secondary Spark is configured — "
                    "run Configure Sparks (enable both Sparks) or move them to primary.")

            # 3. Infra once per job.
            serving.clear_resident_containers(cfg, self.log)
            gr_mod.ensure_grader_image(cfg, self.log)
            serving.ensure_network(cfg, self.log)
            preflight.check_searxng(cfg, self.log)
            hf = bm_config.hf_token()

            # 4. Grade each deck unit, oldest first. One deck's failure never
            #    kills the job.
            self.decks_total = len(units)
            succeeded = 0
            for idx, unit in enumerate(units, 1):
                self.deck_index = idx
                self.company = unit["company_slug"]
                self.period = unit.get("period")
                entry = {"company": unit["company_slug"], "period": unit.get("period"),
                         "status": "running"}
                self.decks.append(entry)
                self.panel = [{"name": r["name"], "model": r["model"], "status": "pending"}
                              for r in valid]
                self._persist()
                try:
                    result = self._grade_deck(cfg, led, job_id, remote_root, unit, idx,
                                              valid, extractor_model, adjudicate, hf)
                    entry.update({"status": "done", "period": result["period"],
                                  "composite": result["composite"]})
                    succeeded += 1
                    comp = result["composite"]
                    self.log(f"[runner] deck done: {unit['company_slug']} {result['period']}"
                             f" composite={comp if comp is not None else '?'}")
                except Exception as e:
                    entry.update({"status": "failed", "error": str(e)[:300]})
                    self.log(f"[runner] DECK FAILED ({unit['company_slug']} "
                             f"{unit.get('period') or '?'}): {e}")
                    self.log(traceback.format_exc().splitlines()[-1])
                self._persist()

            # 5. Job-level report.
            self.phase = "collecting"; self._persist()
            self._write_job_report(job_id)

            # 6. Confidentiality: wipe the deck text from the Spark + teardown.
            if cfg.get("wipeRemoteDocs", True):
                sc.run(sc.head(cfg), f"rm -rf {remote_root}", timeout=120)
                self.log("[runner] wiped deck text from the Spark")
            serving.tear_down_all(cfg, self.log)
            self._live_aliases = None

            # 7. Move the graded originals aside (the company folders stay).
            self._drain_inbox(job_id, units)
            self._last_done_sig = _inbox_signature()[1]

            if succeeded:
                self.phase = "done"
                self.message = (f"Graded {succeeded}/{len(units)} deck(s) with "
                                f"{len(valid)} grader(s).")
            else:
                self.phase = "error"
                self.message = f"All {len(units)} deck(s) failed — see the activity log."
            self.log(f"=== Grading job {job_id} complete ({succeeded}/{len(units)} decks) ===")
            self._persist()
        except Exception as e:
            self.phase = "error"
            self.message = f"Grading failed: {e}"
            self.log(f"[runner] JOB FAILED — {e}")
            self.log(traceback.format_exc().splitlines()[-1])
            # A failed batch counts as seen — otherwise autoRunOnDrop retries
            # the same failing inbox every other tick.
            self._last_done_sig = _inbox_signature()[1]
            try:
                serving.tear_down_all(cfg, self.log)
            except Exception:
                pass
            self._live_aliases = None
            self._persist()

    # ------------------------------------------------------------- one deck
    def _grade_deck(self, cfg: dict, led, job_id: str, remote_root: str, unit: dict,
                    idx: int, panel: list[dict], extractor_model: str,
                    adjudicate: bool, hf) -> dict:
        """Grade one deck unit end to end. Raises on deck failure (caller continues)."""
        slug_c = unit["company_slug"]
        # The staging/remote dir token; the final deck_id is the resolved period.
        token = _token(unit["period"]) if unit.get("period") else f"deck-{idx:02d}"
        local_deck = os.path.join(JOBS_DIR, job_id, slug_c, token)
        remote_deck = f"{remote_root}/{slug_c}/{token}"
        head = sc.head(cfg)

        # --- extract locally + stage + push -----------------------------------
        self.phase = "extracting"; self._persist()
        manifest = extraction.extract_files(unit["files"], os.path.join(local_deck, "docs"),
                                            self.log)
        ok_docs = [m for m in manifest if m["ok"]]
        for m in manifest:
            if not m["ok"]:
                self.log(f"[runner] WARNING: {m['source']}: {m['error']}")
        if not ok_docs:
            raise RuntimeError("no document text could be extracted from this deck")
        gr_mod.stage_deck_files(cfg, local_deck, panel)
        push = sc.push_dir(head, local_deck, remote_deck)
        if push.returncode != 0:
            raise RuntimeError(f"shipping deck text to the Spark failed: {push.stderr}")

        # --- serve in waves; extractor + graders run in their model's wave ----
        self.phase = "grading"; self._persist()
        waves = serving.plan_waves(cfg, {r["model"] for r in panel} | {extractor_model})
        self.waves_total = len(waves)
        for i, wave in enumerate(waves, 1):
            self.wave_index = i
            aliases = {m["alias"] for m in wave}
            wpanel = [r for r in panel if r["model"] in aliases]
            self.log(f"[runner] wave {i}/{len(waves)}: models={sorted(aliases)} "
                     f"graders={[r['name'] for r in wpanel]}"
                     f"{' +extractor' if extractor_model in aliases else ''}")
            # Single-wave jobs keep the wave serving across decks — reloading a
            # 31B from disk between decks costs minutes for the same model set.
            reusable = len(waves) == 1
            reused = reusable and self._live_aliases == aliases
            if reused:
                self.log("[runner] wave already serving from the previous deck — reusing")
            else:
                serving.bring_up_wave(cfg, wave, hf, self.log)
                self._await_serving(cfg, wave)
            try:
                try:
                    preflight.check_wave(cfg, wave, self.log)
                except Exception as e:
                    if not reused:
                        raise
                    # The kept-warm wave went stale (e.g. a vLLM crashed between
                    # decks) — restart it once and re-check.
                    self.log(f"[runner] kept-warm wave failed preflight ({e}); restarting it")
                    self._live_aliases = None
                    serving.tear_down_wave(cfg, wave, self.log)
                    serving.bring_up_wave(cfg, wave, hf, self.log)
                    self._await_serving(cfg, wave)
                    preflight.check_wave(cfg, wave, self.log)
                extract_here = extractor_model in aliases
                if (extract_here and wpanel
                        and _sparks_disjoint(cfg, extractor_model, wpanel)):
                    # Extractor and graders sit on different Sparks (separate
                    # GPUs) — run them concurrently.
                    self.log("[runner] extractor and graders are on different Sparks — "
                             "running them in parallel")
                    holder: dict = {}
                    t = threading.Thread(
                        target=lambda: holder.update(
                            gr_mod.run_extractor(cfg, remote_deck, extractor_model,
                                                 self.log)),
                        daemon=True)
                    t.start()
                    res = gr_mod.run_wave_graders(cfg, remote_deck, wpanel, self.log)
                    self._mark_panel(res)
                    t.join()
                    if not holder.get("report"):
                        raise RuntimeError("extractor produced no extraction.json")
                else:
                    if extract_here:
                        er = gr_mod.run_extractor(cfg, remote_deck, extractor_model,
                                                  self.log)
                        if not er.get("report"):
                            raise RuntimeError("extractor produced no extraction.json")
                    if wpanel:
                        res = gr_mod.run_wave_graders(cfg, remote_deck, wpanel, self.log)
                        self._mark_panel(res)
            finally:
                if reusable:
                    self._live_aliases = aliases  # leave it serving for the next deck
                else:
                    serving.tear_down_wave(cfg, wave, self.log)
                    self._live_aliases = None

        # --- pull the panel outputs + validate --------------------------------
        local_out = os.path.join(local_deck, "out")
        pull = sc.pull_dir(head, f"{remote_deck}/out", local_out)
        if pull.returncode != 0:
            raise RuntimeError(f"pulling panel outputs from the Spark failed: {pull.stderr}")

        ext_path = os.path.join(local_out, "extraction.json")
        ext_obj, ext_err = validate.validate_file(ext_path, "extraction")
        if ext_obj is None or ext_err:
            raise RuntimeError(f"extraction.json invalid: {ext_err or 'missing'}")

        period, deck_id = self._resolve_period(unit, ext_obj, ext_path, token)
        self.period = period; self._persist()

        grades, panel_meta = [], []
        for r in panel:
            gpath = os.path.join(local_out, f"{r['rid']}.json")
            gobj, gerr = (None, "no output file")
            if os.path.exists(gpath) and not os.path.exists(gpath + ".invalid"):
                gobj, gerr = validate.validate_file(gpath, "grades")
            ok = gobj is not None and not gerr
            if ok:
                grades.append(gobj)
            else:
                self.log(f"[runner] grader {r['rid']} report dropped: {gerr}")
            panel_meta.append({"rid": r["rid"], "model": r["model"], "valid": ok})
        if len(grades) < 2:
            raise RuntimeError(f"only {len(grades)} valid grade report(s) (need >= 2)")

        # --- adjudication (non-fatal) ------------------------------------------
        adjudication_md = None
        if adjudicate:
            self.phase = "adjudicating"; self._persist()
            try:
                adj_model = adj_mod.pick_model(cfg)
                for wave in serving.plan_waves(cfg, {adj_model}):
                    if self._live_aliases and \
                            {m["alias"] for m in wave} <= self._live_aliases:
                        # The adjudicator's model is already serving on the
                        # kept-warm wave — use it; cycling the containers here
                        # would tear down the proxy the next deck reuses.
                        self.log("[runner] adjudicator model already serving — reusing wave")
                        adj_mod.run_adjudication(cfg, remote_deck, self.log)
                        continue
                    serving.bring_up_wave(cfg, wave, hf, self.log)
                    try:
                        self._await_serving(cfg, wave)
                        preflight.check_wave(cfg, wave, self.log)
                        adj_mod.run_adjudication(cfg, remote_deck, self.log)
                    finally:
                        # Cycling a different model set invalidates the kept-warm
                        # wave (this replaces/removes the shared proxy).
                        serving.tear_down_wave(cfg, wave, self.log)
                        self._live_aliases = None
                local_adj = os.path.join(local_deck, "adjudicator-out")
                sc.pull_dir(head, f"{remote_deck}/adjudicator-out", local_adj)
                apath = os.path.join(local_adj, "ADJUDICATION.md")
                if os.path.exists(apath):
                    adjudication_md = open(apath, errors="replace").read().strip() or None
                if not adjudication_md:
                    self.log("[runner] WARNING: no adjudication produced (continuing without)")
            except Exception as e:
                self.log(f"[runner] WARNING: adjudication failed (non-fatal): {e}")

        # --- deterministic scoring + ledger ------------------------------------
        self.phase = "scoring"; self._persist()
        company = led.ensure_company(slug_c)
        pinned = company.get("pinned_targets") or []
        aliases_map = company.get("kpi_aliases") or {}
        prior = led.prior_targets(slug_c, period)
        report_deck_dir = os.path.join(REPORTS_DIR, job_id, slug_c, deck_id)
        meta = {
            "company": slug_c,
            "period": period,
            "deck_id": deck_id,
            "job_id": job_id,
            "graded_at": datetime.now(timezone.utc).isoformat(),
            "panel": panel_meta,
            "artifacts": {
                "report_dir": report_deck_dir,
                "extraction": os.path.join(report_deck_dir, "extraction.json"),
                "grades": [os.path.join(report_deck_dir, f"{p['rid']}.json")
                           for p in panel_meta if p["valid"]],
                "adjudication": (os.path.join(report_deck_dir, "ADJUDICATION.md")
                                 if adjudication_md else None),
            },
        }
        record = scoring.score_deck(ext_obj, grades, pinned, prior, aliases_map,
                                    cfg["weights"], meta)
        rec_path = led.record_deck(slug_c, record, ext_obj.get("forward_targets") or [])
        self.log(f"[runner] ledger updated: {rec_path}")

        # --- reports ------------------------------------------------------------
        self.phase = "collecting"; self._persist()
        all_recs = led.deck_records(slug_c)  # period-sorted, incl. the new record
        idx = next((i for i, r in enumerate(all_recs) if r.get("deck_id") == deck_id), -1)
        prev_rec = all_recs[idx - 1] if idx > 0 else None
        deck_md = scorecard.render_deck_report(record, ext_obj, adjudication_md, prev_rec)
        if rec_path and str(rec_path).endswith(".json"):
            ledger_md = os.path.splitext(str(rec_path))[0] + ".md"
        else:
            ledger_md = os.path.join(LEDGER_DIR, slug_c, "decks", f"{deck_id}.md")
        os.makedirs(os.path.dirname(ledger_md), exist_ok=True)
        with open(ledger_md, "w") as f:
            f.write(deck_md)

        os.makedirs(report_deck_dir, exist_ok=True)
        with open(os.path.join(report_deck_dir, "DECK_REPORT.md"), "w") as f:
            f.write(deck_md)
        for fn in sorted(os.listdir(local_out)):  # extraction + raw grader jsons (+ .invalid)
            src = os.path.join(local_out, fn)
            if os.path.isfile(src):
                shutil.copyfile(src, os.path.join(report_deck_dir, fn))
        if adjudication_md:
            with open(os.path.join(report_deck_dir, "ADJUDICATION.md"), "w") as f:
                f.write(adjudication_md + "\n")

        # Refresh the company scorecard + the /data/reports latest copy.
        sc_md = scorecard.render_scorecard(led.get_company(slug_c), led.deck_records(slug_c))
        sc_path = os.path.join(LEDGER_DIR, slug_c, "SCORECARD.md")
        os.makedirs(os.path.dirname(sc_path), exist_ok=True)
        with open(sc_path, "w") as f:
            f.write(sc_md)
        with open(os.path.join(REPORTS_DIR, "latest-scorecard.md"), "w") as f:
            f.write(sc_md)
        self.log(f"[runner] reports saved to {report_deck_dir}")

        return {"period": period, "deck_id": deck_id, "composite": _composite(record),
                "report_dir": report_deck_dir}

    def _resolve_period(self, unit: dict, ext_obj: dict, ext_path: str,
                        token: str) -> tuple[str, str]:
        """(period, deck_id) for this deck. Filename-derived period wins; else the
        extractor's deck.period if canonical-ish; else the file's mtime month
        (flagged as period_inferred in the extraction's red-flag candidates)."""
        if unit.get("period"):
            return unit["period"], _token(unit["period"])
        p = decks.canonicalize_period((ext_obj.get("deck") or {}).get("period"))
        if p:
            self.log(f"[runner] period '{p}' taken from the deck text")
            return p, _token(p)
        try:
            mtime = os.path.getmtime(unit["files"][0])
        except OSError:
            mtime = time.time()
        period = time.strftime("%Y-%m", time.localtime(mtime))
        ext_obj.setdefault("red_flag_candidates", []).append({
            "code": "period_inferred",
            "description": ("Reporting period was not stated in the filename or the deck "
                            f"text; inferred from the file's modification time as {period}."),
            "severity": 2,
            "evidence": "",
        })
        try:
            with open(ext_path, "w") as f:
                json.dump(ext_obj, f, indent=2)
        except Exception:
            pass
        self.log(f"[runner] WARNING: period inferred from file mtime: {period}")
        return period, (_token(period) or token)

    # ------------------------------------------------------------- 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 _write_job_report(self, job_id: str):
        """latest.md — one job summary across every deck graded (or failed)."""
        lines = [f"# Boardroom Map grading job — {job_id}\n"]
        done = [d for d in self.decks if d.get("status") == "done"]
        failed = [d for d in self.decks if d.get("status") == "failed"]
        lines.append(f"Decks graded: {len(done)}/{len(self.decks)}\n")
        lines.append("\n## Results\n")
        for d in self.decks:
            if d.get("status") == "done":
                comp = d.get("composite")
                comp_s = f"{comp:.1f}" if isinstance(comp, (int, float)) else "?"
                lines.append(f"- **{d['company']}** — {d.get('period') or '?'}: "
                             f"composite **{comp_s}** / 100\n")
            else:
                lines.append(f"- **{d['company']}** — {d.get('period') or '?'}: "
                             f"FAILED — {d.get('error', 'unknown error')}\n")
        if done:
            lines.append("\nPer-deck reports (DECK_REPORT.md, extraction, raw grades, "
                         f"adjudication): `/data/reports/{job_id}/<company>/<deck>/`.\n")
            lines.append("Company scorecards: `/data/ledger/<company>/SCORECARD.md` "
                         "(latest copy at `/data/reports/latest-scorecard.md`).\n")
        if failed:
            lines.append("\nFailed decks were still moved to "
                         f"`/data/processed/{job_id}/` — re-drop them to regrade.\n")
        assembled = "".join(lines)
        out_dir = os.path.join(REPORTS_DIR, job_id)
        os.makedirs(out_dir, exist_ok=True)
        with open(os.path.join(out_dir, "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_dir, "report.md")

    def _drain_inbox(self, job_id: str, units: list[dict]):
        """Move each unit's ORIGINAL files to /data/processed/<job>/<slug>/. The
        per-company inbox folders are kept — the user reuses them next quarter."""
        moved = 0
        for unit in units:
            dest = os.path.join(PROCESSED, job_id, unit["company_slug"])
            os.makedirs(dest, exist_ok=True)
            for src in unit["files"]:
                if os.path.isfile(src):
                    try:
                        shutil.move(src, os.path.join(dest, os.path.basename(src)))
                        moved += 1
                    except Exception:
                        pass
        self.log(f"[runner] inbox cleared ({moved} file(s) moved to processed/{job_id}; "
                 "company folders kept)")


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