ui/scoring/observable.py

"""ObservableScorer — the REAL scorer seam: drives the backend observable-detection pipeline.

Implements the same `Scorer.score(parsed) -> ScoreResult` protocol as `DummyScorer`, but the five
axis SCORES come from `prompt_card.observable_pipeline.analyze` (base MiniCPM4.1-8B with the cascade-best
prompts + the locked per-category LoRA hybrid on Modal). Everything else in the UI is unchanged.

What is REAL here: the 5 axis scores, the overall/tier, per-axis confidence (measured Cohen's κ), and the
critical-types the model detected. What stays presentational (reused from the dummy helpers): the evidence
quote selection, the static per-axis tips, and the improvement line — the backend does not surface which
turns triggered each axis, so quotes are chosen heuristically from the real user turns.

PRIVACY: inherits the pipeline's in-memory-only contract — user chat is never written to disk or logged.
Config: needs OPENBMB_BASE_URL / OPENBMB_TOKEN (the scoring endpoint). LoRA routing auto-enables when
`modal` is importable (DISABLE_LORA=1 forces base-only).
"""
from __future__ import annotations

import os

from ..data import AXES
from ..parsing import ParsedExport
from .interface import AxisScore, ScoreResult
from .dummy import _evidence, _improvement, _TIPS, _CRITICAL  # presentational reuse

import os as _os

MAX_CONVS = int(_os.environ.get("UI_MAX_CONVS", "30"))  # representative sample (fast); set 0 for WHOLE history


def _select_convs(conversations, n=MAX_CONVS):
    """Default: a representative even-stride sample of N SUBSTANTIVE conversations (>=2 user turns)
    across the whole export — not first-N (which biases the score) — and EVERY turn within each is
    scored (we sample which conversations, never truncate a conversation). Scores converge by ~20-30
    convs, so this is ~2 min vs ~10+ for the full history at near-identical scores. Set UI_MAX_CONVS=0
    to score the whole history."""
    with_user = [c for c in conversations if any(t.role == "user" for t in c.turns)]
    substantive = [c for c in with_user if sum(1 for t in c.turns if t.role == "user") >= 2]
    pool = substantive or with_user
    if n <= 0 or len(pool) <= n:
        return pool
    step = len(pool) / n                       # even stride across the export (range over time)
    return [pool[int(i * step)] for i in range(n)]

# backend card_data axis name -> UI display name
_AXIS_NAME = {
    "Focus": "Focus",
    "Technique": "Technique",
    "Critical Engagement": "Critical",
    "Interaction": "Interaction",
    "Input Quality": "Input Quality",
}
# backend critical type -> UI CriticalBreakdown key
_CRIT_KEY = {
    "skepticism": "skepticism",
    "source_request": "source_req",
    "rebuttal": "rebuttal",
    "independent_verification": "verify",
    "re_questioning": "re_ask",
}


def _confidence(kappa) -> str:
    """Measured Cohen's κ -> coarse confidence band (None/0 -> low)."""
    if not kappa:
        return "low"
    if kappa >= 0.55:
        return "high"
    if kappa >= 0.40:
        return "medium"
    return "low"


def backend_available() -> bool:
    """True when the scoring endpoint is configured (the minimum to run the real scorer)."""
    return bool(os.environ.get("OPENBMB_BASE_URL") and os.environ.get("OPENBMB_TOKEN"))


def _lora_axes():
    """Enabled per-category LoRA axes. PRODUCTION: set LORA_ENDPOINT=1 when OPENBMB_BASE_URL is the Modal
    vLLM server that serves the adapters by name (HTTP, no `modal` package needed). Else legacy path needs
    `modal` importable. DISABLE_LORA=1 forces base-only everywhere."""
    if os.environ.get("DISABLE_LORA"):
        return set()
    # LoRA only via the warm vLLM ENDPOINT that serves the adapters by name (LORA_ENDPOINT=1 + a
    # base URL pointing at that Modal server). We deliberately do NOT auto-enable just because the
    # `modal` package is importable — that used to trigger the slow per-adapter Modal-job path even
    # against a base-only endpoint, producing the incoherent "LoRA @ shared endpoint" state.
    if os.environ.get("LORA_ENDPOINT") and os.environ.get("OPENBMB_BASE_URL"):
        from prompt_card.llm.lora_router import DEFAULT_ENABLED
        return set(DEFAULT_ENABLED)
    return set()


class ObservableScorer:
    """Real scorer. Implements the `Scorer` protocol; drops into app.py where DummyScorer was."""

    def __init__(self, client=None, embedder=None):
        self._client = client
        self._embedder = embedder

    def _make_client(self):
        from prompt_card.llm.minicpm import MiniCPMClient
        base, token = os.environ.get("OPENBMB_BASE_URL"), os.environ.get("OPENBMB_TOKEN")
        if not base or not token:
            raise RuntimeError("Scoring endpoint not configured (OPENBMB_BASE_URL / OPENBMB_TOKEN).")
        # generous timeout so the FIRST request survives a Modal cold start (~85s, min_containers=0);
        # warm calls are ~1.5s so this only bites on cold/slow. Override via OPENBMB_TIMEOUT.
        # max_tokens cap is the #1 throughput lever: our outputs are tiny JSON, so the 512 default just
        # let the 8B ramble (~5s/call). 128 caps generation to ~1s/call without truncating the JSON.
        return MiniCPMClient(base, token, timeout=int(os.environ.get("OPENBMB_TIMEOUT") or "180"),
                             max_tokens=int(os.environ.get("OPENBMB_MAX_TOKENS") or "128"))

    def _fallback_client(self):
        """Optional secondary endpoint (e.g. the shared OpenBMB free API) used base-only if the
        primary (Modal) endpoint fails. Configured via OPENBMB_FALLBACK_BASE_URL / _TOKEN."""
        base, token = os.environ.get("OPENBMB_FALLBACK_BASE_URL"), os.environ.get("OPENBMB_FALLBACK_TOKEN")
        if not base or not token:
            return None
        from prompt_card.llm.minicpm import MiniCPMClient
        return MiniCPMClient(base, token, max_tokens=int(os.environ.get("OPENBMB_MAX_TOKENS") or "128"))

    def score(self, parsed: ParsedExport, progress=None) -> ScoreResult:
        from prompt_card.observable_pipeline import analyze

        selected = _select_convs(parsed.conversations)
        dicts = [{"turns": [{"role": t.role, "text": t.text} for t in c.turns]} for c in selected]
        if not dicts:
            raise RuntimeError("No user turns to score.")

        client = self._client or self._make_client()
        try:
            data = analyze(dicts, client, self._embedder, name="Player", lora_axes=_lora_axes(),
                           progress=progress)
        except Exception:
            fb = self._fallback_client()
            if fb is None:
                raise   # no fallback -> app.py degrades to DummyScorer
            # primary endpoint (Modal) unavailable -> shared OpenBMB, base-only (no adapters there)
            data = analyze(dicts, fb, self._embedder, name="Player", lora_axes=set(), progress=progress)
        return self._build(data, parsed)

    def _build(self, data, parsed) -> ScoreResult:
        score_by_axis = {a["axis"]: a["score"] for a in data["axes"]}
        conf_by_axis = data.get("axis_confidence", {})
        texts = [t.text for t in parsed.user_turns]
        # real per-type tallies (pipeline now returns counts); fall back to presence for older payloads
        counts = data.get("critical_type_counts")
        present = set(data.get("critical_types_present", []))
        crit_counts = {_CRIT_KEY[k]: (counts.get(k, 0) if counts else (1 if k in present else 0))
                       for k in _CRIT_KEY}

        evidence = data.get("evidence", {})
        axes = []
        for ui_name in AXES:  # canonical UI order
            backend_name = next((b for b, u in _AXIS_NAME.items() if u == ui_name), ui_name)
            score = float(score_by_axis.get(backend_name, 0.0))
            conf = _confidence(conf_by_axis.get(backend_name))
            # REAL evidence = the user turns where this axis fired; fall back to heuristic if none fired
            quotes = [f'"{q}"' for q in evidence.get(ui_name, []) if q] or _evidence(ui_name, texts, crit_counts)
            axes.append(AxisScore(ui_name, score, conf, quotes, _TIPS[ui_name]))

        scores = {a.name: a.score for a in axes}
        return ScoreResult(axes=axes, critical_counts=crit_counts, improvement=_improvement(scores))