diff --git a/src/ui/beta_dashboard.py b/src/ui/beta_dashboard.py index 1121473..56a1698 100644 --- a/src/ui/beta_dashboard.py +++ b/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) diff --git a/src/ui/full_pipeline_elite_status.py b/src/ui/full_pipeline_elite_status.py new file mode 100644 index 0000000..dae6d1a --- /dev/null +++ b/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 diff --git a/src/ui/move_policy.py b/src/ui/move_policy.py index 0c3ff73..2577c53 100644 --- a/src/ui/move_policy.py +++ b/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 diff --git a/tests/test_ui_beta_dashboard.py b/tests/test_ui_beta_dashboard.py index 523c318..0458d47 100644 --- a/tests/test_ui_beta_dashboard.py +++ b/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 diff --git a/tests/test_ui_full_pipeline_elite_status.py b/tests/test_ui_full_pipeline_elite_status.py new file mode 100644 index 0000000..ad6e75e --- /dev/null +++ b/tests/test_ui_full_pipeline_elite_status.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from src.ui.full_pipeline_elite_status import ( + full_pipeline_elite_policy_for_ui, + read_full_pipeline_elite_status, + render_full_pipeline_elite_status, +) +from src.ui.move_policy import ( + BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED, + BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT, + BOT_MOVE_POLICY_STATISTICAL, +) + + +def write_evidence( + root: Path, + *, + username: str = "hikaru", + time_class: str = "rapid", + decision: str = "elite_guarded_ready", + recommended_policy: str = "Direct Ranker Elite Guarded", + failed_checks: list[str] | None = None, +) -> Path: + path = root / username / f"{time_class}_elite_guarded_evidence.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "decision": decision, + "reason": "elite_guarded_evidence_ready", + "recommended_policy": recommended_policy, + "failed_checks": failed_checks or [], + } + ), + encoding="utf-8", + ) + return path + + +def test_read_full_pipeline_elite_status_ready_guarded(tmp_path: Path) -> None: + write_evidence(tmp_path) + + status = read_full_pipeline_elite_status( + username="Hikaru", + time_class="rapid", + evidence_root=tmp_path, + ) + + assert status.ready is True + assert status.technical_error is False + assert status.move_policy == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED + assert status.move_policy_label == "Direct Ranker elite guarded" + assert full_pipeline_elite_policy_for_ui(status) == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED + + +def test_read_full_pipeline_elite_status_ready_strict(tmp_path: Path) -> None: + write_evidence( + tmp_path, + time_class="blitz", + decision="elite_guarded_strict_ready", + recommended_policy="Direct Ranker Elite Guarded Strict", + ) + + status = read_full_pipeline_elite_status( + username="hikaru", + time_class="blitz", + evidence_root=tmp_path, + ) + + assert status.ready is True + assert status.move_policy == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT + assert "strict" in status.move_policy_label.lower() + + +def test_read_full_pipeline_elite_status_needs_more_evidence_is_safe(tmp_path: Path) -> None: + write_evidence( + tmp_path, + decision="needs_more_evidence", + recommended_policy="Direct Ranker Elite Guarded after more evidence", + failed_checks=["runtime.fallback_percent"], + ) + + status = read_full_pipeline_elite_status( + username="hikaru", + time_class="rapid", + evidence_root=tmp_path, + ) + + assert status.ready is False + assert status.move_policy == BOT_MOVE_POLICY_STATISTICAL + assert full_pipeline_elite_policy_for_ui(status) == BOT_MOVE_POLICY_STATISTICAL + assert "runtime.fallback_percent" in status.notes + + +def test_read_full_pipeline_elite_status_missing_is_safe(tmp_path: Path) -> None: + status = read_full_pipeline_elite_status( + username="hikaru", + time_class="rapid", + evidence_root=tmp_path, + ) + + assert status.ready is False + assert status.technical_error is True + assert status.move_policy == BOT_MOVE_POLICY_STATISTICAL + + +def test_render_full_pipeline_elite_status_mentions_ui_policy(tmp_path: Path) -> None: + write_evidence( + tmp_path, + decision="elite_guarded_strict_ready", + recommended_policy="Direct Ranker Elite Guarded Strict", + ) + status = read_full_pipeline_elite_status( + username="hikaru", + time_class="rapid", + evidence_root=tmp_path, + ) + + rendered = render_full_pipeline_elite_status(status) + + assert "Full Pipeline Elite Status" in rendered + assert "Elite Guarded Strict ready" in rendered + assert "Direct Ranker elite guarded strict" in rendered diff --git a/tests/test_ui_move_policy.py b/tests/test_ui_move_policy.py index 7db4b61..c907e54 100644 --- a/tests/test_ui_move_policy.py +++ b/tests/test_ui_move_policy.py @@ -211,3 +211,26 @@ def test_legacy_ml_shadow_still_logs_selector_decision() -> None: assert payload["decision"].uci == "e2e4" assert payload["shadow_result"] == {"enabled": True, "match": True} assert shadow.calls == 1 + +def test_elite_evidence_decision_maps_to_ui_move_policy() -> None: + from src.ui.move_policy import ( + BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED, + BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT, + BOT_MOVE_POLICY_STATISTICAL, + bot_move_policy_for_elite_evidence_decision, + ) + + assert bot_move_policy_for_elite_evidence_decision( + "elite_guarded_ready", + "Direct Ranker Elite Guarded", + ) == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED + + assert bot_move_policy_for_elite_evidence_decision( + "elite_guarded_strict_ready", + "Direct Ranker Elite Guarded Strict", + ) == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT + + assert bot_move_policy_for_elite_evidence_decision( + "needs_more_evidence", + "Direct Ranker Elite Guarded after more evidence", + ) == BOT_MOVE_POLICY_STATISTICAL