diff --git a/src/runtime/readiness.py b/src/runtime/readiness.py index 6b6b4c7..6f5b4c3 100644 --- a/src/runtime/readiness.py +++ b/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,9 +92,6 @@ 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", @@ -100,6 +99,7 @@ def direct_ranker_metrics_candidate_paths( 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)) diff --git a/src/ui/main_ui_elite_recommendation.py b/src/ui/main_ui_elite_recommendation.py index eabaf32..cb30581 100644 --- a/src/ui/main_ui_elite_recommendation.py +++ b/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,15 +55,141 @@ 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, @@ -58,6 +197,18 @@ def build_main_ui_elite_recommendation( 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,6 +330,7 @@ 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( @@ -186,6 +338,7 @@ def apply_main_ui_elite_recommendation_to_window( 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) diff --git a/tests/test_main_ui_elite_recommendation.py b/tests/test_main_ui_elite_recommendation.py index e3a1f98..ce7b941 100644 --- a/tests/test_main_ui_elite_recommendation.py +++ b/tests/test_main_ui_elite_recommendation.py @@ -62,15 +62,44 @@ def write_evidence( return path +def write_blocked_release_gate( + root: Path, + *, + username: str = "hikaru", + time_class: str = "rapid", +) -> Path: + path = root / username / f"{time_class}_release_gate.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "assessment": { + "status": "release_candidate_blocked", + "failed_blocker_checks": 2, + "blocker_check_ids": [ + "runtime.safety.gap_over_200_percent", + "runtime.safety.outside_top_5_percent", + ], + } + } + ), + encoding="utf-8", + ) + return path + + def test_main_ui_elite_recommendation_reads_strict_ready( tmp_path: Path, ) -> None: - write_evidence(tmp_path) + evidence_root = tmp_path / "evidence" + release_gate_root = tmp_path / "release_gate" + write_evidence(evidence_root) recommendation = build_main_ui_elite_recommendation( username="Hikaru", time_class="rapid", - evidence_root=tmp_path, + evidence_root=evidence_root, + release_gate_root=release_gate_root, ) assert recommendation.ready is True @@ -87,13 +116,36 @@ def test_main_ui_elite_recommendation_reads_strict_ready( assert "Move policy key: direct_ranker_elite_guarded_strict" in rendered +def test_main_ui_elite_recommendation_blocks_elite_when_release_gate_blocked( + tmp_path: Path, +) -> None: + evidence_root = tmp_path / "evidence" + release_gate_root = tmp_path / "release_gate" + write_evidence(evidence_root) + write_blocked_release_gate(release_gate_root) + + recommendation = build_main_ui_elite_recommendation( + username="hikaru", + time_class="rapid", + evidence_root=evidence_root, + release_gate_root=release_gate_root, + ) + + assert recommendation.ready is False + assert recommendation.should_select_policy is False + assert recommendation.decision == "release_gate_blocked" + assert recommendation.move_policy == BOT_MOVE_POLICY_STATISTICAL + assert "runtime.safety.gap_over_200_percent" in recommendation.reason + + def test_main_ui_elite_recommendation_missing_evidence_is_safe( tmp_path: Path, ) -> None: recommendation = build_main_ui_elite_recommendation( username="hikaru", time_class="rapid", - evidence_root=tmp_path, + evidence_root=tmp_path / "evidence", + release_gate_root=tmp_path / "release_gate", ) assert recommendation.ready is False @@ -107,7 +159,9 @@ def test_main_ui_elite_recommendation_missing_evidence_is_safe( def test_apply_main_ui_elite_recommendation_selects_combo_policy( tmp_path: Path, ) -> None: - write_evidence(tmp_path) + evidence_root = tmp_path / "evidence" + release_gate_root = tmp_path / "release_gate" + write_evidence(evidence_root) combo = FakeCombo() window = SimpleNamespace( username="hikaru", @@ -125,7 +179,8 @@ def sync() -> None: recommendation = apply_main_ui_elite_recommendation_to_window( window, - evidence_root=tmp_path, + evidence_root=evidence_root, + release_gate_root=release_gate_root, ) assert recommendation.should_select_policy is True @@ -134,11 +189,39 @@ def sync() -> None: assert calls == ["sync"] +def test_apply_main_ui_elite_recommendation_does_not_select_when_release_gate_blocked( + tmp_path: Path, +) -> None: + evidence_root = tmp_path / "evidence" + release_gate_root = tmp_path / "release_gate" + write_evidence(evidence_root) + write_blocked_release_gate(release_gate_root) + combo = FakeCombo() + window = SimpleNamespace( + username="hikaru", + time_class="rapid", + move_policy=BOT_MOVE_POLICY_STATISTICAL, + move_policy_combo=combo, + ) + + recommendation = apply_main_ui_elite_recommendation_to_window( + window, + evidence_root=evidence_root, + release_gate_root=release_gate_root, + ) + + assert recommendation.should_select_policy is False + assert window.move_policy == BOT_MOVE_POLICY_STATISTICAL + assert combo.current_index == 0 + + def test_apply_main_ui_elite_recommendation_does_not_select_when_not_ready( tmp_path: Path, ) -> None: + evidence_root = tmp_path / "evidence" + release_gate_root = tmp_path / "release_gate" path = write_evidence( - tmp_path, + evidence_root, decision="needs_more_evidence", recommended_policy="Direct Ranker Elite Guarded after more evidence", ) @@ -153,6 +236,7 @@ def test_apply_main_ui_elite_recommendation_does_not_select_when_not_ready( recommendation = apply_main_ui_elite_recommendation_to_window( window, evidence_path=path, + release_gate_root=release_gate_root, ) assert recommendation.should_select_policy is False diff --git a/tests/test_runtime_readiness.py b/tests/test_runtime_readiness.py index 09cdad6..98e2af7 100644 --- a/tests/test_runtime_readiness.py +++ b/tests/test_runtime_readiness.py @@ -21,10 +21,12 @@ def write_ready_fixture( username: str = "hikaru", time_class: str = "rapid", decision: str = "elite_guarded_strict_ready", + release_gate_blocked: bool = False, ) -> dict[str, Path]: persona_root = root / "personas" models_root = root / "models" evidence_root = root / "evidence" + release_gate_root = root / "release_gate" engine_path = root / "engines" / "stockfish.exe" persona_path = persona_root / username / f"{time_class}_complete_persona.json" @@ -56,6 +58,25 @@ def write_ready_fixture( encoding="utf-8", ) + release_gate_path = release_gate_root / username / f"{time_class}_release_gate.json" + if release_gate_blocked: + release_gate_path.parent.mkdir(parents=True, exist_ok=True) + release_gate_path.write_text( + json.dumps( + { + "assessment": { + "status": "release_candidate_blocked", + "failed_blocker_checks": 2, + "blocker_check_ids": [ + "runtime.safety.gap_over_200_percent", + "runtime.safety.outside_top_5_percent", + ], + } + } + ), + encoding="utf-8", + ) + engine_path.parent.mkdir(parents=True, exist_ok=True) engine_path.write_text("dummy engine marker", encoding="utf-8") @@ -63,24 +84,38 @@ def write_ready_fixture( "persona_root": persona_root, "models_root": models_root, "evidence_root": evidence_root, + "release_gate_root": release_gate_root, "engine_path": engine_path, "persona_path": persona_path, "model_path": model_path, "metrics_path": metrics_path, "evidence_path": evidence_path, + "release_gate_path": release_gate_path, } -def test_runtime_readiness_ready_for_elite_strict(tmp_path: Path) -> None: - paths = write_ready_fixture(tmp_path) - report = build_runtime_readiness_report( - username="Hikaru", - time_class="rapid", +def build_report_from_paths( + paths: dict[str, Path], + *, + username: str = "hikaru", + time_class: str = "rapid", + strict_metrics: bool = False, +): + return build_runtime_readiness_report( + username=username, + time_class=time_class, engine_path=paths["engine_path"], persona_root=paths["persona_root"], models_root=paths["models_root"], evidence_root=paths["evidence_root"], + release_gate_root=paths["release_gate_root"], + strict_metrics=strict_metrics, ) + + +def test_runtime_readiness_ready_for_elite_strict(tmp_path: Path) -> None: + paths = write_ready_fixture(tmp_path) + report = build_report_from_paths(paths) assert report.ready is True assert report.exit_code == 0 assert report.recommendation.move_policy == BOT_MOVE_POLICY_DIRECT_RANKER_ELITE_GUARDED_STRICT @@ -90,17 +125,25 @@ def test_runtime_readiness_ready_for_elite_strict(tmp_path: Path) -> None: assert "Move policy key: direct_ranker_elite_guarded_strict" in rendered +def test_runtime_readiness_release_gate_blocked_forces_statistical(tmp_path: Path) -> None: + paths = write_ready_fixture(tmp_path, release_gate_blocked=True) + report = build_report_from_paths(paths) + + assert report.ready is False + assert report.exit_code == 2 + assert report.recommendation.move_policy == BOT_MOVE_POLICY_STATISTICAL + assert report.recommendation.decision == "release_gate_blocked" + + rendered = render_runtime_readiness_report(report) + assert "Runtime readiness: NOT READY" in rendered + assert "Elite evidence: release_gate_blocked" in rendered + assert "Move policy key: statistical" in rendered + + def test_runtime_readiness_missing_evidence_blocks_ui_runtime(tmp_path: Path) -> None: paths = write_ready_fixture(tmp_path) paths["evidence_path"].unlink() - report = build_runtime_readiness_report( - username="hikaru", - time_class="rapid", - engine_path=paths["engine_path"], - persona_root=paths["persona_root"], - models_root=paths["models_root"], - evidence_root=paths["evidence_root"], - ) + report = build_report_from_paths(paths) assert report.ready is False assert report.exit_code == 2 assert report.recommendation.move_policy == BOT_MOVE_POLICY_STATISTICAL @@ -111,14 +154,7 @@ def test_runtime_readiness_missing_evidence_blocks_ui_runtime(tmp_path: Path) -> def test_runtime_readiness_needs_more_evidence_blocks(tmp_path: Path) -> None: paths = write_ready_fixture(tmp_path, decision="needs_more_evidence") - report = build_runtime_readiness_report( - username="hikaru", - time_class="rapid", - engine_path=paths["engine_path"], - persona_root=paths["persona_root"], - models_root=paths["models_root"], - evidence_root=paths["evidence_root"], - ) + report = build_report_from_paths(paths) assert report.ready is False assert report.recommendation.move_policy == BOT_MOVE_POLICY_STATISTICAL assert any( @@ -130,14 +166,7 @@ def test_runtime_readiness_needs_more_evidence_blocks(tmp_path: Path) -> None: def test_runtime_readiness_metrics_are_warning_by_default(tmp_path: Path) -> None: paths = write_ready_fixture(tmp_path) paths["metrics_path"].unlink() - report = build_runtime_readiness_report( - username="hikaru", - time_class="rapid", - engine_path=paths["engine_path"], - persona_root=paths["persona_root"], - models_root=paths["models_root"], - evidence_root=paths["evidence_root"], - ) + report = build_report_from_paths(paths) assert report.ready is True metrics_check = next(check for check in report.checks if check.name == "Direct Ranker metrics") assert metrics_check.ok is False @@ -147,13 +176,8 @@ def test_runtime_readiness_metrics_are_warning_by_default(tmp_path: Path) -> Non def test_runtime_readiness_strict_metrics_blocks(tmp_path: Path) -> None: paths = write_ready_fixture(tmp_path) paths["metrics_path"].unlink() - report = build_runtime_readiness_report( - username="hikaru", - time_class="rapid", - engine_path=paths["engine_path"], - persona_root=paths["persona_root"], - models_root=paths["models_root"], - evidence_root=paths["evidence_root"], + report = build_report_from_paths( + paths, strict_metrics=True, ) assert report.ready is False @@ -174,6 +198,7 @@ def test_runtime_readiness_cli_exit_code(tmp_path: Path, capsys) -> None: "persona_root": str(paths["persona_root"]), "models_root": str(paths["models_root"]), "evidence_root": str(paths["evidence_root"]), + "release_gate_root": str(paths["release_gate_root"]), "strict_metrics": False, }, )()