app/app.py

"""CellTriage — QC operator console.

An inference-only interface over the artifacts built by
``src.pipelines.build_app_artifacts``. It loads fitted models from
``outputs/models/`` and a small demo subset; **no raw cycling data is shipped**.

DESIGN PRIORITIES, in order:

1. **The budget advisor**, because Phase 8's flat cost frontier and 99.17%
   chamber-time release is the most actionable result in the project, and a
   slider the user moves themselves is more convincing than a sentence.
2. **The campaign-shift alarm as a STATE CHANGE**, not a disclaimer. Phase 10
   found the model becomes *confidently wrong* under campaign shift — accuracy
   halves while intervals narrow — so a grey footnote is not an adequate
   response to that failure mode. Unfamiliar data changes what the result panel
   looks like.
3. **Decisions rendered visually.** Whether the confidence interval crosses a
   grade boundary is a spatial question; asking a reader to infer it from two
   numbers wastes the result.
4. **Limitations one click away**, never buried.

Run locally:  ``python -m app.app``
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any

# HuggingFace Spaces launches `app_file` AS A SCRIPT -- `python app/app.py` --
# which puts app/ on sys.path instead of the repository root. `import app` then
# resolves to this very file and `from app import panels` fails with a circular
# import, while `python -m app.app` works fine. Local module-mode runs therefore
# cannot detect the one failure mode that matters for deployment, so the repo
# root is put first on the path before any first-party import.
if __package__ in (None, ""):  # pragma: no cover - only on script launch
    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import gradio as gr
import numpy as np
import pandas as pd
from scipy import stats

from app import panels, theme

# ---------------------------------------------------------------------------
# ZeroGPU: why NOTHING here is decorated with @spaces.GPU
#
# This Space runs on ZeroGPU because it is the free tier available for new
# Spaces, not because the workload needs a GPU. Every model is a CPU-bound
# scikit-learn pipeline -- no neural networks anywhere in the project, by hard
# constraint -- and a single prediction takes ~45 ms on CPU.
#
# AN EARLIER VERSION DECORATED THE INFERENCE PATH WITH @spaces.GPU AND THAT WAS
# A REAL BUG, caught only on the deployed Space. `@spaces.GPU` REQUESTS a GPU
# slot per call and bills the declared duration against a daily quota that is
# 5 minutes on a free account. The campaign-shift check alone calls predict()
# once per cell to build the reference distribution -- 22 calls -- so at the
# declared 15 s per call a SINGLE page load reserved 330 s and exhausted the
# entire day's quota. The deployed console failed with "You have exceeded your
# ZeroGPU runs limit" while every local test passed, because locally the
# decorator is documented as effect-free.
#
# SO THE INFERENCE PATH IS NOT DECORATED. What the platform still needs is a
# GPU entry point to exist at all: with no `@spaces.GPU` anywhere the Space sat
# at `hardware: None` against `requested_hardware: zero-a10g`, bound its port,
# and served 503 on every request -- the ZeroGPU container was never
# provisioned. HuggingFace's own instructions list importing `spaces` and
# decorating a GPU function as the steps that make a ZeroGPU Space work.
#
# The resolution is to declare ONE minimal GPU entry point that the console
# never calls on any user path. It satisfies provisioning; it consumes no quota
# because it is never invoked. This is a platform accommodation, stated plainly
# rather than dressed up as a computational need: nothing in this project uses
# a GPU, and no model was changed to pretend otherwise.
try:  # pragma: no cover - the Hub runtime always provides this
    import spaces
except ImportError:  # local development, CI, the test suite
    spaces = None


if spaces is not None:  # pragma: no cover - only meaningful on ZeroGPU

    @spaces.GPU(duration=1)
    def _zerogpu_provisioning_probe() -> str:
        """Declared so ZeroGPU provisions the Space. Never called by the UI.

        Every user-facing path -- prediction, grading, the shift check --
        deliberately avoids this function. See the note above on why decorating
        the real inference path exhausted a 5-minute daily quota in one page
        load.
        """
        return "cpu-bound console; no GPU work is performed"

ROOT = Path(__file__).resolve().parent.parent
ASSETS = ROOT / "app" / "assets"
MODELS = ROOT / "outputs" / "models"
FIGURES = ROOT / "outputs" / "figures"

#: A KS statistic above this against the qualification cohort puts the console
#: into its alarm state. Phase 10 measured D = 0.651 for the shifted campaign
#: and D near zero in-distribution. This threshold is DIRECTIONAL, not
#: calibrated: one shifted campaign can demonstrate the association but cannot
#: establish a decision threshold, and the interface says so when it fires.
KS_ALARM_THRESHOLD = 0.30
ALPHA = 0.10

#: Labels for the incoming-lot selector. The shifted option is real batch-3
#: data, not a perturbation -- the alarm has to be demonstrable on the campaign
#: that actually broke the guarantee, or the console is asserting a safety
#: property it never exercises.
QUALIFICATION_LOT = "Qualification campaigns (batches 1–2)"
SHIFTED_LOT = "New production campaign (batch 3)"

#: The one capacity at which the allocation policies were actually compared
#: (held_fraction = 0.2017 ± 0.0034 over 50 folds, allocation_summary.csv).
EVALUATED_HELD_PERCENT = 20.2


# ---------------------------------------------------------------------------
# Bundle loading (cold start)
# ---------------------------------------------------------------------------


class Bundle:
    """Everything the console needs, loaded once at import."""

    def __init__(self) -> None:
        self.manifest = json.loads((ASSETS / "manifest.json").read_text(encoding="utf-8"))
        self.results = json.loads((ASSETS / "results.json").read_text(encoding="utf-8"))
        self.demo = pd.read_parquet(ASSETS / "demo_cells.parquet")

        self.budgets = [int(b) for b in self.manifest["budgets"]]
        self.grades = self.manifest["grades"]
        self.warranty = int(self.manifest["warranty_target_cycles"])
        self.cost_matrix = self.manifest["cost_matrix"]
        self.grade_order = list(self.grades)
        self.boundaries = {g: float(self.grades[g]["min_cycles"]) for g in self.grade_order}

        # Models are loaded LAZILY and cached. ZeroGPU Spaces spin down
        # aggressively, so cold start matters: the constructor touches only the
        # small metadata bundle (~0.2 MB) and the first prediction for a budget
        # pays for that one model. Loading all five eagerly would add ~6 MB of
        # joblib deserialisation to every cold start for models a given session
        # may never use.
        self._models: dict[int, Any] = {}
        self.features: dict[int, list[str]] = {}
        self.half_widths: dict[int, float] = {}
        for budget in self.budgets:
            entry = self.manifest["models"].get(str(budget))
            if entry and (MODELS / f"triage_extra_trees__budget{budget:03d}.joblib").exists():
                self.features[budget] = entry["feature_names"]
                self.half_widths[budget] = float(entry["conformal_half_width_90"])

        # Campaign membership drives the shift check. Batches 1-2 are the
        # qualification cohort the model and its conformal quantiles were
        # calibrated on; batch 3 is the genuinely shifted production campaign
        # (Phase 3: KS D = 0.651, p = 1.7e-11).
        at_reference = self.demo[self.demo["budget"] == max(self.budgets)]
        self.qualification_ids = sorted(
            at_reference.loc[at_reference["batch"] != "batch3", "cell_id"]
        )
        self.shifted_ids = sorted(
            at_reference.loc[at_reference["batch"] == "batch3", "cell_id"]
        )
        self._reference_cache: dict[int, np.ndarray] = {}

        self.cell_ids = sorted(self.demo["cell_id"].unique())
        self.build_time = self.manifest.get("provenance", {}).get("generated_at_utc", "unknown")

    def model(self, budget: int):
        """Load and cache one budget's model on first use."""
        if budget not in self._models:
            import joblib

            self._models[budget] = joblib.load(
                MODELS / f"triage_extra_trees__budget{budget:03d}.joblib"
            )
        return self._models[budget]

    @property
    def models(self) -> dict[int, Any]:
        """Budgets with a model available (not necessarily loaded yet)."""
        return {b: None for b in self.features}

    def row(self, cell_id: str, budget: int) -> pd.Series:
        subset = self.demo[(self.demo["cell_id"] == cell_id) & (self.demo["budget"] == budget)]
        return subset.iloc[0]

    def predict(self, cell_id: str, budget: int) -> tuple[float, float, float]:
        """Point prediction and conformal interval, in log10 cycle life."""
        row = self.row(cell_id, budget)
        columns = self.features[budget]
        X = pd.DataFrame([[row.get(c, np.nan) for c in columns]], columns=columns)
        centre = float(_infer(self.model(budget), X))
        half = self.half_widths[budget]
        return centre, centre - half, centre + half

    def grade_probabilities(self, lower: float, upper: float) -> dict[str, float]:
        samples = np.linspace(lower, upper, 512)
        cycles = 10.0 ** samples
        ordered = sorted(self.boundaries.items(), key=lambda kv: -kv[1])
        out: dict[str, float] = {}
        previous = np.inf
        for grade, floor in ordered:
            out[grade] = float(((cycles >= floor) & (cycles < previous)).mean())
            previous = floor
        total = sum(out.values()) or 1.0
        return {g: v / total for g, v in out.items()}

    def action_costs(self, probabilities: dict[str, float]) -> dict[str, float]:
        out: dict[str, float] = {}
        for j, assigned in enumerate(self.grade_order):
            out[assigned] = float(sum(
                probabilities[true] * self.cost_matrix[f"true_{true}"][j]
                for true in self.grade_order
            ))
        return out

    def lot_ids(self, campaign: str) -> list[str]:
        """Cell IDs making up an incoming lot from the named campaign."""
        return self.shifted_ids if campaign == SHIFTED_LOT else self.qualification_ids

    def campaign_of(self, cell_id: str) -> str:
        """Which campaign a cell actually came from.

        The lot is a PROPERTY OF THE CELL, never an operator choice. An earlier
        version offered the campaign as a separate control, which let the
        interface show a batch-1 cell under a batch-3 lot -- an incoherent state
        -- and, more seriously, made the shift check something an operator could
        decline to apply. A safety check that the person it protects can switch
        off is not a safety check.
        """
        return SHIFTED_LOT if cell_id in self.shifted_ids else QUALIFICATION_LOT

    def cell_choices(self) -> list[tuple[str, str]]:
        """Dropdown entries labelled with the campaign each cell came from."""
        return [
            (f"{c}  ·  {'batch 3 — new campaign' if c in self.shifted_ids else 'qualification'}", c)
            for c in self.cell_ids
        ]

    def reference_predictions(self, budget: int) -> np.ndarray:
        """Model predictions on the qualification cohort, cached per budget.

        The shift check compares PREDICTIONS on both sides, never true lives.
        A production console has no labels for an incoming lot, so a check that
        needed them would be undeployable -- and comparing true reference lives
        against predicted incoming ones would conflate model bias with
        distribution shift. Both sides go through the same model, so the only
        thing the statistic can move on is the input distribution.
        """
        if budget not in self._reference_cache:
            self._reference_cache[budget] = np.array(
                [self.predict(c, budget)[0] for c in self.qualification_ids]
            )
        return self._reference_cache[budget]

    def grade_for(self, cycles: float) -> str:
        for grade in self.grade_order:
            if cycles >= self.boundaries[grade]:
                return grade
        return self.grade_order[-1]


def _infer(model, X: pd.DataFrame) -> float:
    """The heaviest inference path. Runs on CPU, deliberately undecorated.

    See the ZeroGPU note at the top of this module: decorating this with
    @spaces.GPU exhausted the daily quota on one page load and broke the
    deployed console, because the shift check calls it once per reference cell.
    """
    return float(model.predict(X)[0])


BUNDLE = Bundle()


# ---------------------------------------------------------------------------
# Campaign-shift check
# ---------------------------------------------------------------------------


def campaign_shift_check(campaign: str, budget: int) -> tuple[bool, float, str]:
    """Two-sample KS of the incoming LOT against the qualification cohort.

    Returns (alarm, statistic, html). The alarm is a STATE, not a message: the
    result panel renders differently when it fires, because Phase 10 showed the
    failure mode is a model that becomes confidently wrong, and a footnote is
    not a proportionate response to that.

    THIS IS A LOT-LEVEL PROPERTY AND CANNOT BE OTHERWISE. A single cell carries
    no distribution to test, so an earlier version that passed one prediction to
    a two-sample test could only ever report "not run". Shift is a statement
    about the population a cell arrived in, and the operator screens a cell in
    the context of its lot.

    WHAT THIS CHECK CANNOT DETECT. A shift that leaves the predicted-life
    distribution unchanged while moving the feature-to-life mapping -- same
    marginal, different conditional. That is precisely the regime where the
    model is wrong and the check is silent, and no univariate two-sample test
    on the output can see it.
    """
    lot = BUNDLE.lot_ids(campaign)
    if len(lot) < 3:
        return False, 0.0, _nominal_html(0.0, insufficient=True)

    # Identity is decided on cell IDs, not predicted values. Comparing the float
    # arrays looked equivalent and was not: tree-ensemble reductions differ at
    # ~1e-15 between calls, which is enough for ks_2samp to see two distinct
    # samples and report D = 1/n for a cohort compared against itself.
    if lot == BUNDLE.qualification_ids:
        return False, 0.0, _nominal_html(0.0, is_reference=True, n_lot=len(lot))

    values = np.array([BUNDLE.predict(c, budget)[0] for c in lot])
    reference = BUNDLE.reference_predictions(budget)

    result = stats.ks_2samp(reference, values)
    statistic = float(result.statistic)
    if statistic >= KS_ALARM_THRESHOLD:
        return True, statistic, _alarm_html(statistic, float(result.pvalue), len(lot))
    return False, statistic, _nominal_html(statistic, n_lot=len(lot))


def _alarm_html(statistic: float, pvalue: float, n_lot: int) -> str:
    return f"""
<div class="ct-alarm">
  <div class="ct-alarm-title">⚠ CAMPAIGN SHIFT DETECTED — THE GUARANTEE MAY NOT HOLD</div>
  <p>This lot of {n_lot} cells does not look like the qualification cohort
     (<code>KS D = {statistic:.3f}</code>, p = {pvalue:.2e}, alarm at
     D &ge; {KS_ALARM_THRESHOLD}).</p>
  <p><strong>Conformal validity assumes calibration and production data are
     exchangeable. That assumption is not met here, so the escape-rate bound
     shown below is not guaranteed.</strong></p>
  <p>Measured on the one shifted campaign in this dataset: coverage fell from
     <code>90.7%</code> to <code>42.5%</code> while the prediction interval got
     <code>31.5% NARROWER</code> and error nearly doubled. The model becomes
     <em>confidently wrong</em> — its own confidence signal moves in the
     reassuring direction exactly as it stops being trustworthy.</p>
  <p><strong>Required action:</strong> recalibrate the conformal quantiles on
     labelled cells from this campaign before relying on any decision here.
     Minimum calibration set: 19 cells for &alpha;=0.05, 99 for &alpha;=0.01.</p>
  <p style="color:{theme.TEXT_MUTED};font-size:0.78rem;">The D &ge;
     {KS_ALARM_THRESHOLD} threshold is directional, not calibrated — one shifted
     campaign can demonstrate the association but cannot set a decision
     boundary.</p>
</div>"""


def _nominal_html(
    statistic: float,
    insufficient: bool = False,
    is_reference: bool = False,
    n_lot: int = 0,
) -> str:
    if insufficient:
        return f"""
<div class="ct-nominal">
  <div class="ct-alarm-title">DISTRIBUTION CHECK — NOT RUN</div>
  <p>Fewer than 3 cells in this lot; a two-sample test has nothing to compare.</p>
</div>"""
    if is_reference:
        return f"""
<div class="ct-nominal">
  <div class="ct-alarm-title">✓ DISTRIBUTION CHECK PASSED — REFERENCE LOT</div>
  <p>This cell came from the qualification cohort, so the comparison is against
     its own campaign and passes by construction. That is shown to make the
     contrast legible, <strong>not</strong> as evidence the check works — select
     a cell marked <span class="ct-mono">batch 3 — new campaign</span> to see it
     fire on real out-of-distribution data.</p>
</div>"""
    return f"""
<div class="ct-nominal">
  <div class="ct-alarm-title">✓ DISTRIBUTION CHECK PASSED</div>
  <p>This lot of {n_lot} cells is consistent with the qualification cohort
     (<span class="ct-mono">KS D = {statistic:.3f}</span>, alarm at D &ge;
     {KS_ALARM_THRESHOLD}). The conformal guarantee's exchangeability assumption
     is not contradicted.</p>
</div>"""


# ---------------------------------------------------------------------------
# Tab 1 — screen a cell
# ---------------------------------------------------------------------------


def screen_cell(cell_id: str, budget: int):
    campaign = BUNDLE.campaign_of(cell_id)
    centre, lower, upper = BUNDLE.predict(cell_id, budget)
    probabilities = BUNDLE.grade_probabilities(lower, upper)
    costs = BUNDLE.action_costs(probabilities)
    assigned = min(costs, key=costs.get)

    row = BUNDLE.row(cell_id, budget)
    truth = float(row["cycle_life"])

    escape_probability = float(sum(
        p for g, p in probabilities.items()
        if BUNDLE.boundaries[g] < BUNDLE.warranty
    ))

    alarm, _, shift_html = campaign_shift_check(campaign, budget)

    css_class = {"A": "ct-accept", "B": "ct-continue", "C": "ct-reject"}.get(assigned, "")
    verdict = {"A": "ACCEPT — GRADE A", "B": "ACCEPT — GRADE B", "C": "REJECT — GRADE C"}[assigned]

    # Under alarm the escape figure is still computed and still shown -- hiding
    # it would be its own dishonesty -- but it is displayed as withdrawn rather
    # than as a bound. A green 0.0% sitting under a red shift warning is the
    # exact contradiction Phase 10 is about.
    if alarm:
        escape_cell = (
            f'<div class="ct-v" style="color:{theme.ALARM};">{escape_probability:.1%}'
            f'<span class="ct-u" style="color:{theme.ALARM};">not guaranteed</span></div>'
        )
    else:
        escape_cell = f'<div class="ct-v">{escape_probability:.1%}</div>'

    badge = f"""
<div class="ct-decision {css_class}">
  <div class="ct-label">Triage decision</div>
  <div class="ct-verdict">{verdict}</div>
  <div class="ct-tier">{BUNDLE.grades[assigned]['tier']}</div>
</div>
<div class="ct-readout">
  <div class="ct-cell"><div class="ct-k">Cell</div>
    <div class="ct-v">{cell_id}</div></div>
  <div class="ct-cell"><div class="ct-k">Budget observed</div>
    <div class="ct-v">{budget}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">Predicted life</div>
    <div class="ct-v">{10 ** centre:,.0f}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">90% interval</div>
    <div class="ct-v">{10 ** lower:,.0f}–{10 ** upper:,.0f}</div></div>
  <div class="ct-cell"><div class="ct-k">Escape risk</div>
    {escape_cell}</div>
  <div class="ct-cell"><div class="ct-k">Actual (historical)</div>
    <div class="ct-v">{truth:,.0f}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">Incoming lot</div>
    <div class="ct-v" style="font-size:0.9rem;">{
      'batch 3 — new campaign' if campaign == SHIFTED_LOT else 'qualification'
    }</div></div>
</div>"""

    return (
        badge,
        shift_html,
        panels.decision_plot(10 ** centre, 10 ** lower, 10 ** upper,
                             BUNDLE.boundaries, BUNDLE.warranty, assigned),
        panels.cost_plot(costs, assigned),
        panels.risk_gauge(escape_probability, ALPHA, guaranteed=not alarm),
    )


# ---------------------------------------------------------------------------
# Tab 2 — budget advisor
# ---------------------------------------------------------------------------


def _frontier_costs() -> dict[int, float]:
    scorecard = BUNDLE.results.get("scorecard", [])
    return {int(r["budget"]): float(r["mean"]) for r in scorecard
            if r.get("metric") == "cost_per_cell"}


def budget_advisor(cell_id: str, budget: int):
    costs = _frontier_costs()
    budgets = [b for b in BUNDLE.budgets if b in costs] or BUNDLE.budgets
    widths = [BUNDLE.half_widths[b] * 2 for b in budgets]
    cost_values = [costs.get(b, float("nan")) for b in budgets]

    frontier = BUNDLE.results.get("frontier", {})
    knee = int(frontier.get("knee_budget", budgets[0]))
    released = float(frontier.get("percent_chamber_time_released", float("nan")))

    centre, lower, upper = BUNDLE.predict(cell_id, budget)
    reference_centre, reference_lower, reference_upper = BUNDLE.predict(cell_id, max(budgets))

    span_now = 10 ** upper - 10 ** lower
    span_max = 10 ** reference_upper - 10 ** reference_lower
    saved = 100.0 * (1 - budget / max(budgets))

    summary = f"""
<div class="ct-readout">
  <div class="ct-cell"><div class="ct-k">Budget selected</div>
    <div class="ct-v">{budget}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">Interval width</div>
    <div class="ct-v">{span_now:,.0f}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">At N={max(budgets)}</div>
    <div class="ct-v">{span_max:,.0f}<span class="ct-u">cycles</span></div></div>
  <div class="ct-cell"><div class="ct-k">Expected cost</div>
    <div class="ct-v">{costs.get(budget, float('nan')):.2f}</div></div>
  <div class="ct-cell"><div class="ct-k">Chamber time saved</div>
    <div class="ct-v">{saved:.0f}%<span class="ct-u">vs N={max(budgets)}</span></div></div>
  <div class="ct-cell"><div class="ct-k">Economic knee</div>
    <div class="ct-v">N={knee}</div></div>
</div>
<div class="ct-takeaway">
  <strong>Move the slider and watch the right-hand curve.</strong> The interval
  narrows steadily with budget — more cycles genuinely buy a tighter statement.
  But expected cost is <strong>flat</strong>: every budget from 5 to 100 cycles
  is statistically indistinguishable, so the knee sits at
  <span class="ct-mono">N={knee}</span> and deciding early costs essentially
  nothing. Against cycling every cell to end of life this releases
  <strong>{released:.2f}%</strong> of chamber time.
  <br><br>
  The caveat that must travel with that: the decision is insensitive to budget
  <em>where the prediction is not</em>. Accuracy does improve with more cycles;
  the cost matrix is simply dominated by a few expensive misgrades rather than
  by average accuracy.
</div>"""

    return summary, panels.budget_advisor_plot(budgets, widths, cost_values, budget, knee)


# ---------------------------------------------------------------------------
# Tab 3 — chamber allocation
# ---------------------------------------------------------------------------


def chamber_allocation(slot_percent: float):
    allocation = BUNDLE.results.get("allocation", [])
    costs = {r["policy"]: float(r["mean"]) for r in allocation
             if r.get("metric") == "cost_per_cell"}
    escape = {r["policy"]: float(r["mean"]) for r in allocation
              if r.get("metric") == "escape_rate"}
    if not costs:
        return "<div class='ct-takeaway'>Allocation results unavailable.</div>", None

    order = ["greedy_voi_per_cycle", "random", "uniform", "confidence_only"]
    costs = {k: costs[k] for k in order if k in costs}
    best = min(costs, key=costs.get)

    batch = int(BUNDLE.manifest["batch_size"])
    slots = int(round(batch * slot_percent / 100))

    # The policy comparison was evaluated at ONE capacity, so the slider must
    # not imply it was recomputed at another. Moving off the evaluated point
    # rescales the slot count and nothing else; a control that silently leaves
    # the numbers alone while looking like it changed them is worse than no
    # control at all. Recomputing here is not an option -- it would need the
    # full 50-fold allocation run, and an in-app estimate over 34 demo cells
    # would no longer trace to outputs/reports/allocation_summary.csv.
    off_point = abs(slot_percent - EVALUATED_HELD_PERCENT) > 2.5
    provenance = (
        f'<div class="ct-cell"><div class="ct-k">Cost basis</div>'
        f'<div class="ct-v" style="font-size:0.82rem;color:{theme.CONTINUE};">'
        f'held at {EVALUATED_HELD_PERCENT:.1f}%<span class="ct-u" '
        f'style="color:{theme.CONTINUE};">not recomputed</span></div></div>'
        if off_point else
        f'<div class="ct-cell"><div class="ct-k">Cost basis</div>'
        f'<div class="ct-v" style="font-size:0.82rem;">evaluated point</div></div>'
    )

    caveat = f"""
<div class="ct-alarm" style="border-color:{theme.CONTINUE};
     border-left-color:{theme.CONTINUE};background:rgba(210,153,34,0.08);">
  <div class="ct-alarm-title" style="color:{theme.CONTINUE};">
    SLIDER MOVED OFF THE EVALUATED CAPACITY</div>
  <p>The policy comparison below was measured at a held fraction of
     <code style="color:{theme.CONTINUE};">20.2% ± 0.3</code> across 50 folds and
     <strong>has not been recomputed</strong> at {slot_percent:.0f}%. Only the
     slot count above responds to this control. Treat the ranking as evidence at
     the evaluated capacity, not as a capacity sweep.</p>
</div>""" if off_point else ""

    summary = f"""
<div class="ct-readout">
  <div class="ct-cell"><div class="ct-k">Batch size</div>
    <div class="ct-v">{batch:,}<span class="ct-u">cells</span></div></div>
  <div class="ct-cell"><div class="ct-k">Chamber slots</div>
    <div class="ct-v">{slots:,}</div></div>
  <div class="ct-cell"><div class="ct-k">Best policy</div>
    <div class="ct-v">{costs[best]:.3f}</div></div>
  <div class="ct-cell"><div class="ct-k">vs confidence-only</div>
    <div class="ct-v">{costs.get('confidence_only', 0) - costs[best]:+.3f}</div></div>
  {provenance}
</div>
{caveat}
<div class="ct-takeaway">
  <strong>Uncertainty is not a ranking signal on its own.</strong> Every policy
  here holds the same number of cells and incurs identical chamber cost, so this
  compares only <em>which</em> cells were chosen. Holding the most uncertain
  cells (<span class="ct-mono">confidence_only</span>) is the
  <strong>worst</strong> policy — worse than random — because the most uncertain
  cells are often ones where more testing will not change the decision.
  <br><br>
  Greedy and confidence-only consume identical uncertainty estimates and differ
  only in whether the cost matrix enters the ranking, so the gap between them is
  attributable to the cost matrix specifically. Greedy does <em>not</em>
  significantly beat random at this evaluation's fold size (25/50 folds,
  p = 0.32); a real 1000-cell batch would wash that variance out, but this
  dataset cannot demonstrate it.
</div>"""
    return summary, panels.allocation_plot(costs, escape)


# ---------------------------------------------------------------------------
# Tab 4 — explain
# ---------------------------------------------------------------------------


def explain_cell(cell_id: str, budget: int):
    sheets_path = ROOT / "outputs" / "reports" / "audit_sheets.json"
    drivers: list[dict[str, Any]] = []
    if sheets_path.exists():
        for sheet in json.loads(sheets_path.read_text(encoding="utf-8")):
            if sheet["cell_id"] == cell_id:
                drivers = sheet["drivers"]
                break

    if not drivers:
        return (f"<div class='ct-takeaway'>No stored attribution for "
                f"<span class='ct-mono'>{cell_id}</span>. Attribution sheets are "
                f"precomputed for a representative subset; pick one of: "
                f"{', '.join(_sheet_cells())}.</div>", None, "")

    centre, _, _ = BUNDLE.predict(cell_id, budget)
    figure = panels.shap_waterfall(drivers, base=centre, predicted=centre)

    rows = []
    for i, driver in enumerate(drivers, start=1):
        raw = driver.get("raw_value")
        median = driver.get("cohort_median")
        comparison = ""
        if raw is not None and median is not None:
            comparison = (f"<span class='ct-mono'>{raw:,.4g}</span> "
                          f"(typical <span class='ct-mono'>{median:,.4g}</span> — "
                          f"{'above' if raw > median else 'below'} typical)")
        direction = "raised" if driver["shap"] > 0 else "lowered"
        rows.append(
            f"<li><strong>{driver['description'].capitalize()}</strong><br>"
            f"{comparison}<br>"
            f"<span style='color:{theme.TEXT_MUTED};font-size:0.8rem;'>"
            f"this {direction} the predicted cycle life; a high value means "
            f"{driver['high_means']}</span></li>"
        )

    plain = ("<div class='ct-limit'><h3>Top drivers, in plain language</h3>"
             "<ul>" + "".join(rows) + "</ul></div>")
    takeaway = ("<div class='ct-takeaway'>Values are shown in <strong>physical "
                "units against the cohort median</strong>, not standardised "
                "scores — an audit sheet that reports z-scores is unusable by the "
                "engineer it exists for.</div>")
    return takeaway, figure, plain


def _sheet_cells() -> list[str]:
    path = ROOT / "outputs" / "reports" / "audit_sheets.json"
    if not path.exists():
        return []
    return [s["cell_id"] for s in json.loads(path.read_text(encoding="utf-8"))]


# ---------------------------------------------------------------------------
# Interface
# ---------------------------------------------------------------------------


def _figure(name: str) -> str | None:
    path = FIGURES / name
    return str(path) if path.exists() else None


def _plate(name: str, takeaway_html: str) -> None:
    """Mount a generated figure with its source path stated underneath.

    These are the CANONICAL artifacts -- byte-identical to the files the README
    and docs cite -- rather than dark-theme copies rendered for the console. That
    is a deliberate trade of visual uniformity for traceability: a reviewer can
    check the claim against the exact file that produced it, and a restyled
    duplicate would be one more thing that can silently drift from its source.
    The path caption makes the provenance explicit rather than merely true.
    """
    path = _figure(name)
    if path is None:
        return
    gr.Image(path, label=None, show_label=False, container=False)
    gr.HTML(takeaway_html)
    gr.HTML(f'<div class="ct-source">source · outputs/figures/{name}</div>')


def build_interface() -> gr.Blocks:
    # Gradio 6 moved `theme` and `css` from the Blocks constructor to launch().
    with gr.Blocks(title="CellTriage - QC operator console") as demo:

        gr.HTML(f"""
<div class="ct-masthead">
  <h1>CellTriage · QC OPERATOR CONSOLE</h1>
  <p class="ct-sub">Cost-optimal, risk-controlled end-of-line screening for
     lithium-ion cells — grade assignment with a conformal bound on the escape
     rate.</p>
  <p class="ct-build">model build {BUNDLE.build_time} &nbsp;·&nbsp;
     {len(BUNDLE.models)} budgets &nbsp;·&nbsp; recipe descriptors excluded per
     Phase 10 &nbsp;·&nbsp; inference only, CPU</p>
</div>""")

        with gr.Tabs():
            # ---------------- budget advisor (first: most actionable) --------
            with gr.Tab("Budget advisor"):
                gr.Markdown(
                    "### How many cycles do you actually need?\n"
                    "Aging chambers are the throughput bottleneck. Move the "
                    "slider and watch what more testing buys — and what it does not."
                )
                with gr.Row():
                    ba_cell = gr.Dropdown(BUNDLE.cell_choices(), value=BUNDLE.cell_ids[0],
                                          label="Demo cell", scale=1)
                    # The slider reads in CYCLES, the unit the operator schedules
                    # chamber time in. It snaps to the nearest budget a model was
                    # fitted for; showing a list index instead would make the one
                    # number on the control meaningless.
                    ba_budget = gr.Slider(
                        minimum=min(BUNDLE.budgets), maximum=max(BUNDLE.budgets),
                        step=5, value=max(BUNDLE.budgets),
                        label=f"Diagnostic budget — cycles observed "
                              f"(snaps to {', '.join(str(b) for b in BUNDLE.budgets)})",
                        scale=2)
                ba_summary = gr.HTML()
                ba_plot = gr.Plot()

                def _advise(cell_id, cycles):
                    nearest = min(BUNDLE.budgets, key=lambda b: abs(b - float(cycles)))
                    return budget_advisor(cell_id, nearest)

                for control in (ba_cell, ba_budget):
                    control.change(_advise, [ba_cell, ba_budget], [ba_summary, ba_plot])
                demo.load(_advise, [ba_cell, ba_budget], [ba_summary, ba_plot])

            # ---------------- screen a cell ---------------------------------
            with gr.Tab("Screen a cell"):
                gr.Markdown(
                    "### Screen one cell\n"
                    "Cells marked *batch 3 — new campaign* are real "
                    "out-of-distribution data, not a simulated shift. Select one "
                    "to see the console change state."
                )
                with gr.Row():
                    sc_cell = gr.Dropdown(BUNDLE.cell_choices(), value=BUNDLE.cell_ids[0],
                                          label="Cell", scale=1)
                    sc_budget = gr.Radio([str(b) for b in BUNDLE.budgets],
                                         value=str(BUNDLE.budgets[-1]),
                                         label="Cycles observed", scale=2)
                sc_badge = gr.HTML()
                sc_shift = gr.HTML()
                sc_decision = gr.Plot()
                with gr.Row():
                    sc_cost = gr.Plot()
                    sc_risk = gr.Plot()

                def _screen(cell_id, budget):
                    return screen_cell(cell_id, int(budget))

                sc_inputs = [sc_cell, sc_budget]
                sc_outputs = [sc_badge, sc_shift, sc_decision, sc_cost, sc_risk]
                for control in sc_inputs:
                    control.change(_screen, sc_inputs, sc_outputs)
                demo.load(_screen, sc_inputs, sc_outputs)

            # ---------------- chamber allocation ----------------------------
            with gr.Tab("Chamber allocation"):
                gr.Markdown(
                    "### Which cells deserve the scarce slots?\n"
                    "You cannot hold every cell. Given a fixed number of chamber "
                    "slots, the question is which cells to hold."
                )
                ca_slots = gr.Slider(
                    5, 40, value=20, step=5,
                    label=f"Chamber slots as % of batch — policies were compared "
                          f"at {EVALUATED_HELD_PERCENT}% only")
                ca_summary = gr.HTML()
                ca_plot = gr.Plot()
                ca_slots.change(chamber_allocation, ca_slots, [ca_summary, ca_plot])
                demo.load(chamber_allocation, ca_slots, [ca_summary, ca_plot])

            # ---------------- explain ---------------------------------------
            with gr.Tab("Explain this decision"):
                gr.Markdown(
                    "### Why did the system decide that?\n"
                    "A scrap decision must be defensible to a process engineer "
                    "who does not read SHAP plots."
                )
                ex_cell = gr.Dropdown(_sheet_cells() or BUNDLE.cell_ids,
                                      value=(_sheet_cells() or BUNDLE.cell_ids)[0],
                                      label="Cell (attribution precomputed)")
                ex_note = gr.HTML()
                ex_plot = gr.Plot()
                ex_plain = gr.HTML()

                def _explain(cell_id):
                    return explain_cell(cell_id, BUNDLE.budgets[-1])

                ex_cell.change(_explain, ex_cell, [ex_note, ex_plot, ex_plain])
                demo.load(_explain, ex_cell, [ex_note, ex_plot, ex_plain])

            # ---------------- the science ------------------------------------
            with gr.Tab("The science"):
                gr.Markdown("### The physical basis, and the operating frontier")
                gr.HTML(f"""
<div class="ct-takeaway" style="border-left-color:{theme.TEXT_DIM};">
  <strong style="color:{theme.TEXT_MUTED};">Why these figures are light against a
  dark console.</strong> They are the <em>canonical</em> artifacts —
  byte-identical to the files the README and <span class="ct-mono">docs/</span>
  cite, and each states its source path below. Rendering dark duplicates for the
  console would look tidier and give a reviewer a second copy that can silently
  drift from the one the claims were made against. Traceability was judged worth
  more than visual uniformity.
</div>""")

                _plate("fig05_dqv_variance_canary.png", """<div class="ct-takeaway">
                  <strong>This one plot is why early prediction works.</strong>
                  The variance of ΔQ(V) — how the discharge curve's shape changes
                  between cycles 10 and 100 — predicts cycle life at
                  <span class="ct-mono">R² = 0.859</span> before any meaningful
                  capacity fade is visible. Reproducing the published
                  relationship (ρ = −0.93) was the gate every later result
                  depended on.</div>""")

                _plate("fig15_budget_frontier.png", """<div class="ct-takeaway">
                  <strong>Cost is flat across the budget range; escape rate is
                  not.</strong> The left panel shows expected cost per cell
                  against the baselines it must beat; the right is the Pareto
                  frontier of escape rate against diagnostic cost. Down and left
                  is better.</div>""")

                _plate("fig13_conformal_coverage_and_width.png", """<div class="ct-takeaway">
                  <strong>The guarantee holds in-distribution and fails under
                  campaign shift.</strong> Solid lines sit on the diagonal;
                  dashed lines — a new production campaign — fall far below it,
                  in places beneath the uncalibrated baseline the method exists
                  to improve on.</div>""")

            # ---------------- method and limitations -------------------------
            with gr.Tab("Method & limitations"):
                gr.HTML(f"""
<div class="ct-limit">
  <h3>Read this first</h3>
  <ul>
    <li><strong>n = 124 cells.</strong> This is a small dataset for the number of
        questions asked of it. Every headline number is reported as mean ± std
        across 50 outer cross-validation folds with a bootstrap 95% CI; a single
        test-set number would be misleading at this sample size.</li>
    <li><strong>The guarantee does not survive a new production campaign.</strong>
        Conformal coverage fell from 90.7% to <span class="ct-mono">42.5%</span>
        on a later campaign, while the prediction interval got
        <span class="ct-mono">31.5% narrower</span> and error nearly doubled. The
        model becomes <em>confidently wrong</em>. Recalibration on the new
        campaign is required before the escape bound means anything.</li>
    <li><strong>This is a research cycling dataset, not a factory dataset.</strong>
        Cells were cycled in a temperature-controlled laboratory at 30 °C, not
        produced and screened on a line. The QC framing — each cell a unit at
        end-of-line, each charging protocol a process recipe, each batch a
        production campaign — is a faithful analogue, not a literal production
        log.</li>
    <li><strong>Costs are relative units, not currency.</strong> Only ratios
        between entries carry meaning. Every conclusion is tested across an
        escape:overkill sweep from 2:1 to 500:1.</li>
    <li><strong>The advantage over a well-tuned static threshold is
        conditional.</strong> It appears above roughly 6:1 escape:overkill and is
        unfavourable below it. Against classical AQL lot acceptance sampling the
        advantage is unconditional (~14×, 50/50 folds).</li>
    <li><strong>α = 0.01 is unreachable</strong> at this sample size: the
        finite-sample conformal correction needs 99 calibration cells and this
        cohort cannot supply them alongside a training set.</li>
    <li><strong>Grade A has 11 cells.</strong> Grading uses ordinal regression
        from predicted cycle life rather than three-class classification for that
        reason.</li>
  </ul>
</div>
<div class="ct-limit">
  <h3>Method in brief</h3>
  <ul>
    <li>Features use only cycles 1..N, enforced structurally and verified by a
        mutation test that deliberately weakens the budget slice.</li>
    <li>Extra trees over 48 features; recipe descriptors deliberately
        <em>excluded</em> — they cost 1.3% in-distribution and 33 coverage points
        under campaign shift.</li>
    <li>Split-conformal intervals at 90%; decisions minimise expected cost under
        a 3×3 grade cost matrix in which escapes dominate overkill 10.9:1.</li>
    <li>Reproduces Severson et al. (2019) at 12.42% mean percent error against a
        published 9.1%, under the published evaluation design.</li>
  </ul>
</div>""")

        gr.HTML(f"""<p class="ct-build" style="margin-top:14px;">
          CellTriage · inference-only console · no raw cycling data is shipped ·
          demo subset of {len(BUNDLE.cell_ids)} cells</p>""")

    return demo


def main() -> None:
    """Launch the console.

    Theme and CSS are passed to launch() rather than to Blocks: Gradio 6 moved
    them there, and `show_api` was removed in the same release.

    DELIBERATELY PLAIN. Successive attempts to out-guess the Spaces runner --
    exposing a module-level `demo`, wrapping `launch` to re-inject styling,
    ceding the port, forcing `block_thread()` -- each fixed an imagined problem
    and left the Space unprovisioned at `hardware: None`. The structure below is
    the one that actually reached RUNNING on ZeroGPU with the theme intact.
    """
    build_interface().launch(
        server_name="0.0.0.0",
        server_port=7860,
        theme=theme.build_theme(),
        css=theme.CSS,
    )


if __name__ == "__main__":
    main()