diff --git a/orchestrator/bm_config.py b/orchestrator/bm_config.py index abc4c3c..330f110 100644 --- a/orchestrator/bm_config.py +++ b/orchestrator/bm_config.py @@ -51,9 +51,10 @@ CONFIG_DEFAULTS = { "models": [ {"alias": "grader-a", "hfModel": "Qwen/Qwen3-32B-FP8", "spark": "primary", "port": 8001}, ], - # Grading panel + # Grading panel (>= 2 required: every deck needs >= 2 valid grade reports) "graders": [ {"name": "munger-lens", "model": "grader-a", "persona": "", "temperature": None}, + {"name": "girdley-operator", "model": "grader-a", "persona": "", "temperature": None}, ], # Which catalog model runs the stage-1 structured extractor ("" = first model) "extractorModel": "", diff --git a/orchestrator/decks.py b/orchestrator/decks.py index c783367..68528a6 100644 --- a/orchestrator/decks.py +++ b/orchestrator/decks.py @@ -12,7 +12,7 @@ from __future__ import annotations import os import re -SUPPORTED_EXTS = {".pdf", ".pptx", ".docx", ".txt", ".md", ".text"} +SUPPORTED_EXTS = {".pdf", ".pptx", ".docx", ".txt", ".md", ".markdown", ".text"} # Canonical period forms: "2026-Q2", "2026-H1", "2026-05", "FY2026". _PERIOD_PATTERNS = [ @@ -55,6 +55,22 @@ def parse_period_from_name(filename: str) -> str | None: return None +def canonicalize_period(text: str | None) -> str | None: + """Normalize a free-form period string ("2026-q2", "FY 2026", "2026/05") + to the canonical form, or None. Extractor-supplied periods and target + periods must pass through here so ledger lookups match filename periods.""" + if not text: + return None + s = str(text).strip().replace("/", "-") + if period_sort_key(s) != _UNKNOWN_KEY: + return s # already canonical + for pat, canon in _PERIOD_PATTERNS: + m = pat.search(s) + if m: + return canon(m) + return None + + def period_sort_key(period: str | None) -> tuple: """(year, start_month, granularity_rank); unknown/None sorts last.""" if not period: diff --git a/orchestrator/graders.py b/orchestrator/graders.py index e66372d..dbb6de7 100644 --- a/orchestrator/graders.py +++ b/orchestrator/graders.py @@ -83,6 +83,9 @@ def roster(cfg: dict) -> list[dict]: for w in (cfg.get("graders") or []): name = (w.get("name") or "grader").strip() rid = slug(name) + if rid in ("extractor", "adjudicator"): + # reserved role ids: personas/containers/outputs would collide + rid = f"{rid}-grader" if rid in seen: seen[rid] += 1 rid = f"{rid}-{seen[rid]}" diff --git a/orchestrator/jobs.py b/orchestrator/jobs.py index 432f309..c17f652 100644 --- a/orchestrator/jobs.py +++ b/orchestrator/jobs.py @@ -70,11 +70,6 @@ REQUEST_PATH = os.path.join(STATE_DIR, "run_request") TICK_SECONDS = 10 RUNNING_PHASES = ("extracting", "grading", "adjudicating", "scoring", "collecting") -# Canonical-ish reporting periods: 2026-Q2, 2026-H1, FY2026, 2026-05, 2026. -_PERIOD_RE = re.compile( - r"^(?:FY\s?-?\d{4}|\d{4}(?:[-/ ]?(?:Q[1-4]|H[12]|0[1-9]|1[0-2]))?)$", re.IGNORECASE) - - 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).""" @@ -229,8 +224,13 @@ class JobRunner: 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 grading") + 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: @@ -288,6 +288,10 @@ class JobRunner: 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()) # Air-gapped mode can't route to second-Spark models (internal net). @@ -364,6 +368,9 @@ class JobRunner: 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: @@ -545,8 +552,8 @@ class JobRunner: (flagged as period_inferred in the extraction's red-flag candidates).""" if unit.get("period"): return unit["period"], _token(unit["period"]) - p = ((ext_obj.get("deck") or {}).get("period") or "").strip() - if p and _PERIOD_RE.match(p): + 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: diff --git a/orchestrator/ledger.py b/orchestrator/ledger.py index 9f7255a..dbeb377 100644 --- a/orchestrator/ledger.py +++ b/orchestrator/ledger.py @@ -109,6 +109,7 @@ class Ledger: def prior_targets(self, slug: str, period: str) -> list[dict]: """Forward targets an earlier deck set for `period` (this deck's exam).""" company = self.get_company(slug) + period = decks_mod.canonicalize_period(period) if not company or not period: return [] return (company.get("extracted_targets", {}).get(period) or {}).get("targets", []) @@ -143,7 +144,9 @@ class Ledger: from_key = decks_mod.period_sort_key(period) by_period: dict[str, list[dict]] = {} for ft in forward_targets or []: - tp = ft.get("target_period") + # Extractor-supplied target periods are free-form ("2026-q3"); + # canonicalize so the next deck's filename period finds them. + tp = decks_mod.canonicalize_period(ft.get("target_period")) if tp: by_period.setdefault(tp, []).append(ft) for tp, targets in by_period.items(): diff --git a/orchestrator/scoring.py b/orchestrator/scoring.py index fedb7ac..f1c4289 100644 --- a/orchestrator/scoring.py +++ b/orchestrator/scoring.py @@ -87,7 +87,8 @@ def _resolve_target(kpi: dict, pinned_targets: list[dict], prior_targets: list[d cand, via = match_kpi(canon, pinned_cands, aliases) if cand is not None: p = cand["_src"] - return {"target": p.get("target"), "direction": p.get("direction") or kpi.get("direction")}, "pinned", via + return {"target": p.get("target"), "direction": p.get("direction") or kpi.get("direction"), + "profitability": p.get("profitability")}, "pinned", via cand, via = match_kpi(canon, prior_targets or [], aliases) if cand is not None: return {"target": cand.get("target"), @@ -166,10 +167,15 @@ def score_deck(extraction: dict, grades: list[dict], pinned_targets: list[dict], target = float(tgt["target"]) credit = _credit(float(k.get("actual", 0)), target, tgt.get("direction") or k.get("direction") or "gte", floor) + # The operator's pinned classification beats the extractor's guess — + # a pinned profitability KPI must score in the profitability bucket. + profitability = bool(k.get("profitability")) + if tgt is not None and tgt.get("profitability") is not None: + profitability = bool(tgt["profitability"]) kpi_results.append({ "canonical_name": k.get("canonical_name"), "name": k.get("name"), "actual": k.get("actual"), "unit": k.get("unit") or "", - "direction": k.get("direction"), "profitability": bool(k.get("profitability")), + "direction": k.get("direction"), "profitability": profitability, "target": target, "target_source": source, "matched_via": via, "credit": None if credit is None else round(credit, 4), }) diff --git a/orchestrator/serving.py b/orchestrator/serving.py index 6e379c1..75134c2 100644 --- a/orchestrator/serving.py +++ b/orchestrator/serving.py @@ -167,6 +167,7 @@ def tear_down_all(cfg: dict, log) -> None: """Best-effort: remove every Boardroom Map serving container + the network.""" for sp in sc.sparks(cfg): sc.run(sp, "docker ps -aq --filter name=bm-vllm- | xargs -r docker rm -f; " + "docker ps -aq --filter name=bm-grader- | xargs -r docker rm -f; " f"docker rm -f {PROXY_NAME} 2>/dev/null; true", timeout=120) remove_network(cfg, log) log("[serving] all serving torn down") diff --git a/orchestrator/templates/index.html b/orchestrator/templates/index.html index e935093..3eb7d41 100644 --- a/orchestrator/templates/index.html +++ b/orchestrator/templates/index.html @@ -490,7 +490,7 @@ async function refresh(){ const deckChips=(rt.decks||[]).map(d=>{ const st=d.status||'pending'; - const cls=st==='done'?'done':st==='error'?'error':(st==='pending'?'':'busy'); + const cls=st==='done'?'done':(st==='error'||st==='failed')?'error':(st==='pending'?'':'busy'); const tail=d.composite!=null?(' · '+(+d.composite).toFixed(1)):(d.error?' · error':''); return `${esc(d.company)}${d.period?' '+esc(d.period):''}${tail}`; }).join(''); diff --git a/startos/file-models/config.ts b/startos/file-models/config.ts index 4ecec67..ee75912 100644 --- a/startos/file-models/config.ts +++ b/startos/file-models/config.ts @@ -86,7 +86,9 @@ export const configShape = z.object({ }), ) .default([ + // >= 2 graders required: every deck needs >= 2 valid grade reports. { name: 'munger-lens', model: 'grader-a', persona: '', temperature: null }, + { name: 'girdley-operator', model: 'grader-a', persona: '', temperature: null }, ]), // Which catalog model runs the stage-1 structured KPI extractor over each