Read full-pipeline elite status in UI

src/ui/beta_dashboard.py

@@ -338,3 +338,53 @@ def render_beta_dashboard(state: BetaDashboardState | None) -> str:
         f"Last audit: {state.last_audit_status}",
     ]
     return "\n".join(lines)
+
+# Step 32G: render full-pipeline elite evidence status in UI/dashboard text.
+def render_full_pipeline_elite_status_panel(status: Any) -> str:
+    data = _as_dict(status)
+    get_value = data.get if data else lambda key, default=None: getattr(status, key, default)
+
+    decision = str(get_value("decision", "") or get_value("status", "") or "-")
+    reason = str(get_value("reason", "-") or "-")
+    recommended_policy = str(get_value("recommended_policy", "-") or "-")
+    move_policy_label = str(
+        get_value("move_policy_label", "")
+        or get_value("policy_label", "")
+        or get_value("move_policy", "")
+        or "-"
+    )
+    evidence_path = str(
+        get_value("evidence_path", "")
+        or get_value("release_gate_path", "")
+        or "-"
+    )
+
+    ready = decision in {"elite_guarded_ready", "elite_guarded_strict_ready"}
+
+    if decision == "elite_guarded_ready":
+        title = "Elite Guarded ready"
+    elif decision == "elite_guarded_strict_ready":
+        title = "Elite Guarded Strict ready"
+    elif decision == "needs_more_evidence":
+        title = "Elite evidence needs more data"
+    elif decision == "elite_guarded_blocked":
+        title = "Elite guarded runtime blocked"
+    else:
+        title = "Elite evidence unavailable"
+
+    lines = [
+        "Full Pipeline Elite Status",
+        f"Status: {title}",
+        f"Decision: {decision}",
+        f"Reason: {reason}",
+        f"Recommended policy: {recommended_policy}",
+        f"UI move policy: {move_policy_label}",
+        f"Evidence report: {evidence_path}",
+    ]
+
+    if ready:
+        lines.append("UI can enable elite model-first guarded runtime.")
+    else:
+        lines.append("UI should keep elite model-first guarded runtime disabled.")
+
+    return "\n".join(lines)

src/ui/full_pipeline_elite_status.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from src.ui.move_policy import (
+    BOT_MOVE_POLICY_STATISTICAL,
+    bot_move_policy_for_elite_evidence_decision,
+    bot_move_policy_label,
+)
+
+DEFAULT_ELITE_EVIDENCE_ROOT = Path("data/ml/direct_ranker_elite_guarded_evidence")
+ELITE_READY_DECISIONS = {"elite_guarded_ready", "elite_guarded_strict_ready"}
+
+
+@dataclass(frozen=True)
+class UIElitePipelineStatus:
+    username: str
+    time_class: str
+    status: str
+    decision: str
+    reason: str
+    recommended_policy: str
+    move_policy: str
+    move_policy_label: str
+    evidence_path: Path
+    ready: bool
+    technical_error: bool = False
+    notes: tuple[str, ...] = ()
+
+
+def _clean(value: object) -> str:
+    return str(value or "").strip().lower()
+
+
+def elite_evidence_path_for_ui(
+    *,
+    username: object,
+    time_class: object,
+    evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
+) -> Path:
+    return Path(evidence_root) / _clean(username) / f"{_clean(time_class)}_elite_guarded_evidence.json"
+
+
+def _load_json_object(path: Path) -> dict[str, Any] | None:
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
+def _text(payload: dict[str, Any], *keys: str) -> str:
+    for key in keys:
+        value = payload.get(key)
+        if value is not None and str(value).strip():
+            return str(value).strip()
+    return ""
+
+
+def _notes(payload: dict[str, Any]) -> tuple[str, ...]:
+    items: list[str] = []
+    for key in ("failed_checks", "notes", "details", "blocking_failures", "failures"):
+        value = payload.get(key)
+        if isinstance(value, list):
+            items.extend(str(item) for item in value)
+        elif isinstance(value, str):
+            items.append(value)
+    deduped: list[str] = []
+    for item in items:
+        clean = item.strip()
+        if clean and clean not in deduped:
+            deduped.append(clean)
+    return tuple(deduped)
+
+
+def read_full_pipeline_elite_status(
+    *,
+    username: object,
+    time_class: object,
+    evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
+    evidence_path: Path | None = None,
+) -> UIElitePipelineStatus:
+    clean_username = _clean(username)
+    clean_time_class = _clean(time_class)
+    path = evidence_path or elite_evidence_path_for_ui(
+        username=clean_username,
+        time_class=clean_time_class,
+        evidence_root=evidence_root,
+    )
+    payload = _load_json_object(path)
+
+    if payload is None:
+        move_policy = BOT_MOVE_POLICY_STATISTICAL
+        return UIElitePipelineStatus(
+            username=clean_username,
+            time_class=clean_time_class,
+            status="elite_evidence_missing",
+            decision="elite_evidence_missing",
+            reason="elite_evidence_report_unreadable_or_missing",
+            recommended_policy="Statistical selector",
+            move_policy=move_policy,
+            move_policy_label=bot_move_policy_label(move_policy),
+            evidence_path=path,
+            ready=False,
+            technical_error=True,
+            notes=(f"Elite evidence report tidak bisa dibaca: {path}",),
+        )
+
+    decision = _text(payload, "decision", "status") or "unknown_elite_evidence_decision"
+    reason = _text(payload, "reason") or "unknown"
+    recommended_policy = _text(payload, "recommended_policy", "policy_label") or "Statistical selector"
+    move_policy = bot_move_policy_for_elite_evidence_decision(decision, recommended_policy)
+
+    return UIElitePipelineStatus(
+        username=clean_username,
+        time_class=clean_time_class,
+        status=decision,
+        decision=decision,
+        reason=reason,
+        recommended_policy=recommended_policy,
+        move_policy=move_policy,
+        move_policy_label=bot_move_policy_label(move_policy),
+        evidence_path=path,
+        ready=decision in ELITE_READY_DECISIONS,
+        technical_error=False,
+        notes=_notes(payload),
+    )
+
+
+def is_full_pipeline_elite_ready(status: UIElitePipelineStatus) -> bool:
+    return bool(status.ready and not status.technical_error)
+
+
+def full_pipeline_elite_policy_for_ui(status: UIElitePipelineStatus) -> str:
+    if is_full_pipeline_elite_ready(status):
+        return status.move_policy
+    return BOT_MOVE_POLICY_STATISTICAL
+
+
+def render_full_pipeline_elite_status(status: UIElitePipelineStatus) -> str:
+    if status.decision == "elite_guarded_ready":
+        title = "Elite Guarded ready"
+    elif status.decision == "elite_guarded_strict_ready":
+        title = "Elite Guarded Strict ready"
+    elif status.decision == "needs_more_evidence":
+        title = "Elite evidence needs more data"
+    elif status.decision == "elite_guarded_blocked":
+        title = "Elite guarded runtime blocked"
+    else:
+        title = "Elite evidence unavailable"
+
+    lines = [
+        "Full Pipeline Elite Status",
+        f"Status: {title}",
+        f"Username: {status.username}",
+        f"Time class: {status.time_class}",
+        f"Decision: {status.decision}",
+        f"Reason: {status.reason}",
+        f"Recommended policy: {status.recommended_policy}",
+        f"UI move policy: {status.move_policy_label}",
+        f"Evidence report: {status.evidence_path}",
+    ]
+    if status.ready:
+        lines.append("UI can enable this elite model-first guarded policy.")
+    else:
+        lines.append("UI should keep elite model-first guarded policy disabled until evidence is ready.")
+    if status.notes:
+        lines.append("Details:")
+        for note in status.notes:
+            lines.append(f"- {note}")
+    return "\n".join(lines)
+
+
+read_ui_full_pipeline_elite_status = read_full_pipeline_elite_status
+render_ui_full_pipeline_elite_status = render_full_pipeline_elite_status

src/ui/move_policy.py

@@ -79,3 +79,21 @@ def bot_move_policy_label(
             return label
 
     return "Statistical selector"
+
+# Step 32G: map full-pipeline elite evidence decisions to UI move policies.
+def bot_move_policy_for_elite_evidence_decision(
+    decision: object,
+    recommended_policy: object = "",
+) -> str:
+    clean_decision = str(decision or "").strip().lower()
+    clean_recommended = str(recommended_policy or "").strip().lower()
+
+    if clean_decision == "elite_guarded_strict_ready":
+        return BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT
+
+    if clean_decision == "elite_guarded_ready":
+        if "strict" in clean_recommended:
+            return BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT
+        return BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED
+
+    return BOT_MOVE_POLICY_STATISTICAL

tests/test_ui_beta_dashboard.py

@@ -70,3 +70,21 @@ def test_dashboard_records_shadow_turn():
     )
     assert state.shadow_logged == 1
     assert "Last ML move: d4" in render_beta_dashboard(state)
+
+def test_beta_dashboard_renders_full_pipeline_elite_status_panel() -> None:
+    from src.ui.beta_dashboard import render_full_pipeline_elite_status_panel
+
+    rendered = render_full_pipeline_elite_status_panel(
+        {
+            "decision": "elite_guarded_strict_ready",
+            "reason": "elite_guarded_evidence_ready",
+            "recommended_policy": "Direct Ranker Elite Guarded Strict",
+            "move_policy_label": "Direct Ranker elite guarded strict",
+            "evidence_path": "data/ml/direct_ranker_elite_guarded_evidence/hikaru/rapid_elite_guarded_evidence.json",
+        }
+    )
+
+    assert "Full Pipeline Elite Status" in rendered
+    assert "Elite Guarded Strict ready" in rendered
+    assert "Direct Ranker elite guarded strict" in rendered
+    assert "UI can enable" in rendered