From 506e6c79bd3e2d10e69a92013f50295a370c3ce3 Mon Sep 17 00:00:00 2001 From: Jonathan Kirkwood Date: Fri, 31 Jul 2026 15:01:10 -0500 Subject: [PATCH] v0.1.8: two-Spark pipeline + kept-warm serving - Secondary-Spark models now work in air-gapped mode: the LiteLLM proxy is dual-homed onto the default bridge (docker network connect) to reach the secondary's published vLLM port; grader containers stay on the --internal network with zero egress. The head-only enforcement is replaced by a secondary-configured check. - Extraction runs in parallel with grading when the extractor's model and every grader model in the wave sit on different Sparks (separate GPUs). - Keep-warm: single-wave jobs no longer tear the wave down between decks (was a ~6-min 31B reload per deck); a kept-warm wave that fails preflight is restarted once. Adjudicator reuses the live wave when its model is already serving instead of cycling the shared proxy. - clear_resident_containers (preJobStopContainers) now stops names on every configured Spark; health() reports containers on both Sparks. - Verified with a mocked dry-run of the full job loop (3 decks: one bring-up, zero mid-job teardowns, parallel overlap, stale-wave restart). Co-Authored-By: Claude Fable 5 --- README.md | 8 +- orchestrator/jobs.py | 105 ++++++++++++++++++++++----- orchestrator/serving.py | 39 +++++++--- startos/actions/configure-grading.ts | 9 ++- startos/file-models/config.ts | 20 +++-- startos/versions/index.ts | 5 +- startos/versions/v_0_1_8.ts | 18 +++++ 7 files changed, 158 insertions(+), 46 deletions(-) create mode 100644 startos/versions/v_0_1_8.ts diff --git a/README.md b/README.md index 90d196a..b1eab98 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,7 @@ Canonical repo: `https://gitea.ten31.ai/Ten31AI/boardroom-map`. ## Status -**v0.1.7 — live in production.** Deployed on a StartOS box driving a DGX Spark +**v0.1.8 — live in production.** Deployed on a StartOS box driving a DGX Spark in single-spark air-gapped mode (gemma-4-31B panel: munger-lens / girdley-operator / buffett-owner). First full grading run completed 2026-07-29: a three-deck company history graded end-to-end into a running @@ -138,7 +138,11 @@ dashboard, an at-a-glance strip per deck report, and a concise "At a glance" summary table atop every generated DECK_REPORT.md; v0.1.7 added company deletion from the dashboard (wipes the graded history, optionally restores the graded deck files to the inbox) so an evaluation can be re-run from -scratch with a different model panel. +scratch with a different model panel; v0.1.8 added the two-Spark pipeline — +secondary-Spark models work in air-gapped mode (dual-homed proxy; graders +keep zero egress), extraction and grading run in parallel when their models +sit on different Sparks, and single-wave jobs keep the vLLMs warm across +decks instead of reloading the 31B (~6 min) per deck. Known optimization not yet done: the wave is torn down per deck, so the 31B reloads from disk (~6 min) between decks even when the model set is unchanged. diff --git a/orchestrator/jobs.py b/orchestrator/jobs.py index fe9c597..02a93be 100644 --- a/orchestrator/jobs.py +++ b/orchestrator/jobs.py @@ -94,6 +94,15 @@ def _token(s: str) -> str: 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): @@ -125,6 +134,7 @@ class JobRunner: 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 @@ -250,6 +260,7 @@ class JobRunner: 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 ===") @@ -294,15 +305,16 @@ class JobRunner: 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). - if cfg.get("networkMode") == "airgapped": - cat = {m["alias"]: m for m in models} - offenders = [a for a in needed_all 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.") + # 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) @@ -350,6 +362,7 @@ class JobRunner: 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) @@ -376,6 +389,7 @@ class JobRunner: serving.tear_down_all(cfg, self.log) except Exception: pass + self._live_aliases = None self._persist() # ------------------------------------------------------------- one deck @@ -416,19 +430,63 @@ class JobRunner: 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 ''}") - serving.bring_up_wave(cfg, wave, hf, self.log) - try: + # 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) - preflight.check_wave(cfg, wave, self.log) - if extractor_model in aliases: - 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: + 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: - serving.tear_down_wave(cfg, wave, self.log) + 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") @@ -466,13 +524,24 @@ class JobRunner: 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") diff --git a/orchestrator/serving.py b/orchestrator/serving.py index 1411092..088f280 100644 --- a/orchestrator/serving.py +++ b/orchestrator/serving.py @@ -15,9 +15,11 @@ Network topology (the confidentiality boundary): * Head-Spark vLLMs join this network; the proxy reaches them by container name (bm-vllm-). Reviewers reach the proxy by name (boardroom-proxy). * Second-Spark vLLMs publish a host port; the proxy reaches them over the LAN. - This only works in local_services mode (an --internal network can't route to - the LAN), so airgapped jobs must keep all models on the head Spark — enforced - in preflight. + In airgapped mode the per-job network is --internal (no LAN route), so the + proxy gets a second leg on the default bridge (docker network connect) — + the proxy can then reach the secondary Spark while the grader containers + stay internal-only with zero egress. Panel traffic to a secondary-Spark + model crosses the LAN between the two Sparks in plaintext HTTP. Models are served from a pre-populated HF cache mounted from the Spark work dir, so airgapped serving needs no live download. @@ -67,18 +69,19 @@ def remove_network(cfg: dict, log) -> None: # ---------------------------------------------------------- resident models def clear_resident_containers(cfg: dict, log) -> None: - """Stop the user-listed containers on the head Spark so a grading job gets - the GPU to itself (both Sparks normally run an always-on 31B vLLM). + """Stop the user-listed containers on EVERY configured Spark so a grading + job gets the GPUs to itself (both Sparks normally run an always-on 31B + vLLM). Names that don't exist on a given Spark are silently skipped. Deliberately NOT restarted after the job: whatever owns them is responsible for bringing them back (the Gazette's Fleet job reloads its own models).""" names = (cfg.get("preJobStopContainers") or "").replace(",", " ").split() if not names: return - head = sc.head(cfg) quoted = " ".join(shlex.quote(n) for n in names) - log(f"[serving] freeing the GPU on {head.host}: docker stop {' '.join(names)}") - sc.run(head, f"docker stop {quoted} 2>/dev/null; true", timeout=180) + for sp in sc.sparks(cfg): + log(f"[serving] freeing the GPU on {sp.host}: docker stop {' '.join(names)}") + sc.run(sp, f"docker stop {quoted} 2>/dev/null; true", timeout=180) # ------------------------------------------------------------------ vLLM @@ -173,6 +176,15 @@ def bring_up_wave(cfg: dict, wave: list[dict], hf_token: str | None, log) -> Non if r.returncode != 0: raise RuntimeError(f"LiteLLM router launch failed: {r.stderr or r.stdout}") + if cfg.get("networkMode") == "airgapped" and \ + any(m.get("spark") == "secondary" for m in wave): + # The --internal per-job network can't route to the LAN. Give the proxy + # a second leg on the default bridge so it can reach the secondary + # Spark's published vLLM port. The grader containers stay internal-only. + r = sc.run(head, f"docker network connect bridge {PROXY_NAME} 2>/dev/null; true", + timeout=30) + log("[serving] proxy dual-homed onto the bridge (secondary-Spark routing)") + def tear_down_wave(cfg: dict, wave: list[dict], log) -> None: names = " ".join(_vllm_name(m["alias"]) for m in wave) @@ -226,7 +238,10 @@ def reviewer_proxy_base(cfg: dict) -> str: def health(cfg: dict) -> dict: - head = sc.head(cfg) - r = sc.run(head, "docker ps --filter name=bm-vllm- --filter name=boardroom-proxy " - "--format '{{.Names}} {{.Status}}'", timeout=30) - return {"running": (r.stdout or "").strip().splitlines()} + """Serving containers across ALL configured Sparks (vLLMs may sit on either).""" + running: list[str] = [] + for sp in sc.sparks(cfg): + r = sc.run(sp, "docker ps --filter name=bm-vllm- --filter name=boardroom-proxy " + "--format '{{.Names}} {{.Status}}'", timeout=30) + running += (r.stdout or "").strip().splitlines() + return {"running": running} diff --git a/startos/actions/configure-grading.ts b/startos/actions/configure-grading.ts index aef4f06..dbf3b69 100644 --- a/startos/actions/configure-grading.ts +++ b/startos/actions/configure-grading.ts @@ -29,10 +29,11 @@ const inputSpec = InputSpec.of({ name: 'Network Mode', description: 'Air-gapped: graders reach ONLY the on-Spark model proxy — zero internet, ' + - 'board decks never leave your hardware (models must be pre-pulled into the ' + - 'Spark HF cache, all on the head Spark). Local services: graders may also ' + - 'reach LAN services like SearXNG and the second Spark (this network has ' + - 'egress unless you firewall it).', + 'board decks never leave your hardware (models must be pre-pulled into ' + + 'each Spark HF cache; models on either Spark are fine — the proxy routes ' + + 'to the second Spark over your LAN). Local services: graders may also ' + + 'reach LAN services like SearXNG (this network has egress unless you ' + + 'firewall it).', default: 'airgapped', values: { airgapped: 'Air-gapped (no network, recommended)', diff --git a/startos/file-models/config.ts b/startos/file-models/config.ts index 2b78176..1edd0ea 100644 --- a/startos/file-models/config.ts +++ b/startos/file-models/config.ts @@ -61,8 +61,10 @@ export const configShape = z.object({ z.object({ alias: z.string(), hfModel: z.string(), - // Which Spark serves this model. In `airgapped` network mode all models - // must be on the head Spark (see networkMode). + // Which Spark serves this model. Secondary-Spark models work in BOTH + // network modes (in airgapped mode the proxy is dual-homed onto the + // bridge to reach them; graders stay on the internal network). Panel + // traffic to a secondary model crosses the LAN between the Sparks. spark: z.enum(['primary', 'secondary']).default('primary'), port: z.number().int().positive().default(8001), }), @@ -144,9 +146,10 @@ export const configShape = z.object({ // Confidentiality posture for the grader containers: // 'airgapped' — graders join an --internal Docker network: they can // reach ONLY the on-Spark model proxy, with zero internet - // egress. Models must be pre-pulled into the Spark's HF - // cache (no live download). All models must be on the head - // Spark. Strongest confidentiality. + // egress. Models must be pre-pulled into each Spark's HF + // cache (no live download). Secondary-Spark models are + // reached by the dual-homed proxy over the Spark-to-Spark + // LAN. Strongest confidentiality. // 'local_services' — graders may also reach configured LAN services // (e.g. SearXNG) and the second Spark. NOTE: this network // has egress unless you firewall it — use only when you @@ -172,9 +175,10 @@ export const configShape = z.object({ autoRunOnDrop: z.boolean().default(false), // Name of the per-job Docker network created on the head Spark. networkName: z.string().default('boardroom-net'), - // Space/comma-separated docker container names stopped on the head Spark at - // the start of every job to free GPU memory (e.g. an always-on vLLM another - // service runs). NOT restarted afterwards — their owner reloads them. + // Space/comma-separated docker container names stopped on EVERY configured + // Spark at the start of every job to free GPU memory (e.g. an always-on vLLM + // another service runs); names absent on a Spark are skipped. NOT restarted + // afterwards — their owner reloads them. preJobStopContainers: z.string().default(''), // --- Portfolio companies --- diff --git a/startos/versions/index.ts b/startos/versions/index.ts index 5685e72..ce85400 100644 --- a/startos/versions/index.ts +++ b/startos/versions/index.ts @@ -7,9 +7,10 @@ import { v_0_1_4 } from './v_0_1_4' import { v_0_1_5 } from './v_0_1_5' import { v_0_1_6 } from './v_0_1_6' import { v_0_1_7 } from './v_0_1_7' +import { v_0_1_8 } from './v_0_1_8' /** The current version MUST be the first argument (`current`). */ export const versions = VersionGraph.of({ - current: v_0_1_7, - other: [v_0_1_6, v_0_1_5, v_0_1_4, v_0_1_3, v_0_1_2, v_0_1_1, v_0_1_0], + current: v_0_1_8, + other: [v_0_1_7, v_0_1_6, v_0_1_5, v_0_1_4, v_0_1_3, v_0_1_2, v_0_1_1, v_0_1_0], }) diff --git a/startos/versions/v_0_1_8.ts b/startos/versions/v_0_1_8.ts new file mode 100644 index 0000000..bdd1b6f --- /dev/null +++ b/startos/versions/v_0_1_8.ts @@ -0,0 +1,18 @@ +import { VersionInfo } from '@start9labs/start-sdk' + +/** Two-Spark pipeline + kept-warm serving. ExVer form `:`. */ +export const v_0_1_8 = VersionInfo.of({ + version: '0.1.8:0', + releaseNotes: + 'Two-Spark grading pipeline: models pinned to the secondary Spark now work ' + + 'in air-gapped mode too (the model proxy is dual-homed onto the bridge to ' + + 'reach them; grader containers keep zero egress), so the extractor can run ' + + 'on one Spark while the grading panel runs on the other — and when they sit ' + + 'on different Sparks, extraction and grading of a deck run in parallel. ' + + 'Serving is also kept warm across decks: single-wave jobs no longer tear ' + + 'the vLLMs down between decks (previously a ~6-minute 31B reload per deck), ' + + 'with an automatic restart if a kept-warm model fails preflight. ' + + 'preJobStopContainers now stops resident containers on every configured ' + + 'Spark, and Serving/Job health shows containers on both Sparks.', + migrations: {}, +})