Block elite policy when release gate has blockers

src/runtime/readiness.py

@@ -7,6 +7,7 @@
 
 from src.engine.stockfish_service import DEFAULT_ENGINE_PATH
 from src.ui.main_ui_elite_recommendation import (
+    DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT,
     MainUIEliteRecommendation,
     build_main_ui_elite_recommendation,
 )
@@ -15,6 +16,7 @@
 DEFAULT_PERSONA_ROOT = Path("data/personas")
 DEFAULT_DIRECT_MODELS_ROOT = Path("data/ml/direct_candidate_models")
 DEFAULT_ELITE_EVIDENCE_ROOT = Path("data/ml/direct_ranker_elite_guarded_evidence")
+DEFAULT_RELEASE_GATE_ROOT = DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT
 
 
 @dataclass(frozen=True)
@@ -90,16 +92,14 @@ def direct_ranker_metrics_candidate_paths(
     tc = normalize_runtime_name(time_class)
     root = Path(models_root)
 
-    # Important: keep this scoped to models_root.
-    # Do not read global data/ml release-gate files here because isolated tests
-    # can accidentally pass when local artifacts exist for the same username.
     return (
         root / user / f"{tc}_direct_ranker_metrics.json",
         root / user / f"{tc}_metrics.json",
         root / user / f"{tc}_direct_ranker_model_metrics.json",
         root / user / f"{tc}_training_metrics.json",
     )
 
+
 def _path_check(
     *,
     name: str,
@@ -250,6 +250,7 @@ def build_runtime_readiness_report(
     persona_root: Path | str = DEFAULT_PERSONA_ROOT,
     models_root: Path | str = DEFAULT_DIRECT_MODELS_ROOT,
     evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
+    release_gate_root: Path | str = DEFAULT_RELEASE_GATE_ROOT,
     strict_metrics: bool = False,
 ) -> RuntimeReadinessReport:
     user = normalize_runtime_name(username)
@@ -258,6 +259,7 @@ def build_runtime_readiness_report(
         username=user,
         time_class=tc,
         evidence_root=evidence_root,
+        release_gate_root=release_gate_root,
     )
     checks = (
         _persona_check(
@@ -332,6 +334,7 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--persona-root", default=str(DEFAULT_PERSONA_ROOT))
     parser.add_argument("--models-root", default=str(DEFAULT_DIRECT_MODELS_ROOT))
     parser.add_argument("--evidence-root", default=str(DEFAULT_ELITE_EVIDENCE_ROOT))
+    parser.add_argument("--release-gate-root", default=str(DEFAULT_RELEASE_GATE_ROOT))
     parser.add_argument(
         "--strict-metrics",
         action="store_true",
@@ -348,6 +351,7 @@ def run_from_args(args: argparse.Namespace) -> int:
         persona_root=args.persona_root,
         models_root=args.models_root,
         evidence_root=args.evidence_root,
+        release_gate_root=args.release_gate_root,
         strict_metrics=bool(args.strict_metrics),
     )
     print(render_runtime_readiness_report(report))

src/ui/main_ui_elite_recommendation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -17,6 +18,9 @@
 )
 
 
+DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT = Path("data/ml/direct_candidate_release_gate")
+
+
 @dataclass(frozen=True)
 class MainUIEliteRecommendation:
     username: str
@@ -33,6 +37,15 @@ class MainUIEliteRecommendation:
     should_select_policy: bool
 
 
+@dataclass(frozen=True)
+class DirectRankerReleaseGateBlock:
+    path: Path
+    blocked: bool
+    status: str
+    blocker_check_ids: tuple[str, ...]
+    reason: str
+
+
 def _title_for_decision(decision: str) -> str:
     if decision == "elite_guarded_ready":
         return "Elite Guarded ready"
@@ -42,22 +55,160 @@ def _title_for_decision(decision: str) -> str:
         return "Elite evidence needs more data"
     if decision == "elite_guarded_blocked":
         return "Elite guarded runtime blocked"
+    if decision == "release_gate_blocked":
+        return "Direct Ranker release gate blocked"
     return "Elite evidence unavailable"
 
 
+def _normalize_name(value: object) -> str:
+    return str(value or "").strip().lower()
+
+
+def direct_ranker_release_gate_path(
+    *,
+    username: object,
+    time_class: object,
+    release_gate_root: Path | str = DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT,
+) -> Path:
+    user = _normalize_name(username)
+    tc = _normalize_name(time_class)
+    return Path(release_gate_root) / user / f"{tc}_release_gate.json"
+
+
+def read_direct_ranker_release_gate_block(
+    *,
+    username: object,
+    time_class: object,
+    release_gate_root: Path | str = DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT,
+) -> DirectRankerReleaseGateBlock:
+    path = direct_ranker_release_gate_path(
+        username=username,
+        time_class=time_class,
+        release_gate_root=release_gate_root,
+    )
+
+    if not path.exists():
+        return DirectRankerReleaseGateBlock(
+            path=path,
+            blocked=False,
+            status="release_gate_missing",
+            blocker_check_ids=(),
+            reason="release_gate_report_missing",
+        )
+
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        return DirectRankerReleaseGateBlock(
+            path=path,
+            blocked=True,
+            status="release_gate_unreadable",
+            blocker_check_ids=(),
+            reason=f"release_gate_report_unreadable: {exc}",
+        )
+
+    assessment = payload.get("assessment", {})
+    if not isinstance(assessment, dict):
+        assessment = {}
+
+    status = str(
+        assessment.get("status")
+        or payload.get("status")
+        or ""
+    ).strip().lower()
+
+    raw_blockers = assessment.get("blocker_check_ids", [])
+    if not isinstance(raw_blockers, list):
+        raw_blockers = []
+
+    blocker_check_ids = tuple(
+        str(item)
+        for item in raw_blockers
+        if str(item).strip()
+    )
+
+    failed_blocker_checks = assessment.get(
+        "failed_blocker_checks",
+        0,
+    )
+    try:
+        failed_blocker_count = int(failed_blocker_checks)
+    except (TypeError, ValueError):
+        failed_blocker_count = 0
+
+    blocked = (
+        status == "release_candidate_blocked"
+        or failed_blocker_count > 0
+        or bool(blocker_check_ids)
+    )
+
+    if blocked:
+        reason = "release_gate_blocked"
+        if status:
+            reason += f": {status}"
+        if blocker_check_ids:
+            reason += " | blockers: " + ", ".join(blocker_check_ids[:8])
+    else:
+        reason = status or "release_gate_not_blocking"
+
+    return DirectRankerReleaseGateBlock(
+        path=path,
+        blocked=blocked,
+        status=status or "unknown",
+        blocker_check_ids=blocker_check_ids,
+        reason=reason,
+    )
+
+
+def _blocked_by_release_gate_recommendation(
+    *,
+    status: UIElitePipelineStatus,
+    release_gate_block: DirectRankerReleaseGateBlock,
+) -> MainUIEliteRecommendation:
+    policy = BOT_MOVE_POLICY_STATISTICAL
+
+    return MainUIEliteRecommendation(
+        username=status.username,
+        time_class=status.time_class,
+        status=status.status,
+        decision="release_gate_blocked",
+        reason=release_gate_block.reason,
+        recommended_policy="Statistical selector",
+        move_policy=policy,
+        move_policy_label=bot_move_policy_label(policy),
+        evidence_path=status.evidence_path,
+        ready=False,
+        technical_error=False,
+        should_select_policy=False,
+    )
+
+
 def build_main_ui_elite_recommendation(
     *,
     username: object,
     time_class: object,
     evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
     evidence_path: Path | None = None,
+    release_gate_root: Path | str = DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT,
 ) -> MainUIEliteRecommendation:
     status = read_full_pipeline_elite_status(
         username=username,
         time_class=time_class,
         evidence_root=evidence_root,
         evidence_path=evidence_path,
     )
+    release_gate_block = read_direct_ranker_release_gate_block(
+        username=status.username,
+        time_class=status.time_class,
+        release_gate_root=release_gate_root,
+    )
+
+    if status.ready and release_gate_block.blocked:
+        return _blocked_by_release_gate_recommendation(
+            status=status,
+            release_gate_block=release_gate_block,
+        )
+
     policy = full_pipeline_elite_policy_for_ui(status)
     normalized_policy = normalize_bot_move_policy(policy)
 
@@ -179,13 +330,15 @@ def apply_main_ui_elite_recommendation_to_window(
     time_class: object | None = None,
     evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
     evidence_path: Path | None = None,
+    release_gate_root: Path | str = DEFAULT_DIRECT_RANKER_RELEASE_GATE_ROOT,
     auto_select_ready: bool = True,
 ) -> MainUIEliteRecommendation:
     recommendation = build_main_ui_elite_recommendation(
         username=username if username is not None else getattr(window, "username", ""),
         time_class=time_class if time_class is not None else getattr(window, "time_class", ""),
         evidence_root=evidence_root,
         evidence_path=evidence_path,
+        release_gate_root=release_gate_root,
     )
 
     setattr(window, "main_ui_elite_recommendation", recommendation)