diff --git a/evaluate_direct_ranker_elite_guarded_evidence.py b/evaluate_direct_ranker_elite_guarded_evidence.py new file mode 100644 index 0000000..425b70c --- /dev/null +++ b/evaluate_direct_ranker_elite_guarded_evidence.py @@ -0,0 +1,5 @@ +from src.ml.direct_ranker_elite_guarded_evidence import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ml/direct_ranker_elite_guarded_evidence.py b/src/ml/direct_ranker_elite_guarded_evidence.py new file mode 100644 index 0000000..1dc02af --- /dev/null +++ b/src/ml/direct_ranker_elite_guarded_evidence.py @@ -0,0 +1,947 @@ +from __future__ import annotations + +import argparse +import json +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping, Sequence + + +DIRECT_RANKER_ELITE_GUARDED = "direct_ranker_elite_guarded" +DIRECT_RANKER_ELITE_GUARDED_STRICT = "direct_ranker_elite_guarded_strict" +ELITE_GUARDED_POLICIES = { + DIRECT_RANKER_ELITE_GUARDED, + DIRECT_RANKER_ELITE_GUARDED_STRICT, +} + +DEFAULT_SELFPLAY_ROOT = Path("data/ml/direct_ranker_selfplay") +DEFAULT_OUTPUT_ROOT = Path("data/ml/direct_ranker_elite_guarded_evidence") + +ELITE_EVIDENCE_SCHEMA_VERSION = 1 + + +@dataclass(frozen=True) +class EliteEvidenceThresholds: + minimum_decisions: int = 320 + minimum_games: int = 8 + minimum_phase_records: int = 20 + max_fallback_percent: float = 50.0 + strict_max_fallback_percent: float = 70.0 + block_on_invalid_lines: bool = True + + +@dataclass(frozen=True) +class EliteEvidenceResult: + schema_version: int + generated_at: str + username: str + time_class: str + move_policy: str + decision: str + reason: str + recommended_policy: str + selfplay_summary_path: str + output_root: str + thresholds: dict[str, Any] + metrics: dict[str, Any] + failed_checks: tuple[str, ...] + notes: tuple[str, ...] + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def normalize_policy(value: object) -> str: + return str(value or "").strip().lower() + + +def is_elite_policy(value: object) -> bool: + return normalize_policy(value) in ELITE_GUARDED_POLICIES + + +def expected_ready_decision(policy: object) -> str: + normalized = normalize_policy(policy) + if normalized == DIRECT_RANKER_ELITE_GUARDED_STRICT: + return "elite_guarded_strict_ready" + return "elite_guarded_ready" + + +def recommended_policy_for_decision( + *, + decision: str, + move_policy: str, +) -> str: + if decision == "elite_guarded_ready": + return "Direct Ranker Elite Guarded" + if decision == "elite_guarded_strict_ready": + return "Direct Ranker Elite Guarded Strict" + if move_policy == DIRECT_RANKER_ELITE_GUARDED_STRICT: + return "Direct Ranker Elite Guarded Strict after more evidence" + return "Direct Ranker Elite Guarded after more evidence" + + +def selfplay_summary_path( + *, + username: str, + time_class: str, + root: Path | str = DEFAULT_SELFPLAY_ROOT, +) -> Path: + return ( + Path(root) + / username.strip().lower() + / time_class.strip().lower() + / "selfplay_summary.json" + ) + + +def evidence_output_paths( + *, + username: str, + time_class: str, + output_root: Path | str = DEFAULT_OUTPUT_ROOT, +) -> tuple[Path, Path]: + directory = Path(output_root) / username.strip().lower() + stem = f"{time_class.strip().lower()}_elite_guarded_evidence" + return directory / f"{stem}.json", directory / f"{stem}.md" + + +def _as_mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _as_sequence(value: Any) -> Sequence[Any]: + if isinstance(value, (list, tuple)): + return value + return () + + +def _safe_int(value: Any, default: int = 0) -> int: + if value is None: + return default + try: + return int(value) + except (TypeError, ValueError): + try: + return int(float(value)) + except (TypeError, ValueError): + return default + + +def _safe_float(value: Any, default: float = 0.0) -> float: + if value is None: + return default + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _deep_find_first(data: Any, keys: Sequence[str]) -> Any: + if isinstance(data, Mapping): + for key in keys: + if key in data: + return data[key] + for value in data.values(): + found = _deep_find_first(value, keys) + if found is not None: + return found + elif isinstance(data, list): + for item in data: + found = _deep_find_first(item, keys) + if found is not None: + return found + return None + + +def _deep_find_mapping(data: Any, keys: Sequence[str]) -> Mapping[str, Any]: + found = _deep_find_first(data, keys) + return _as_mapping(found) + + +def _phase_counts(summary: Mapping[str, Any]) -> dict[str, int]: + phase_payload = _deep_find_mapping(summary, ("phase_counts", "phases", "phase_records")) + counts = { + "opening": _safe_int( + phase_payload.get("opening") + or _deep_find_first(summary, ("opening", "opening_records")) + ), + "middlegame": _safe_int( + phase_payload.get("middlegame") + or _deep_find_first(summary, ("middlegame", "middlegame_records")) + ), + "endgame": _safe_int( + phase_payload.get("endgame") + or _deep_find_first(summary, ("endgame", "endgame_records")) + ), + } + return counts + + +def _fallback_reasons(summary: Mapping[str, Any]) -> dict[str, int]: + raw = _deep_find_mapping( + summary, + ( + "guarded_fallback_reasons", + "fallback_reasons", + "elite_fallback_reasons", + ), + ) + return {str(key): _safe_int(value) for key, value in raw.items()} + + +def extract_elite_selfplay_metrics( + summary: Mapping[str, Any], +) -> dict[str, Any]: + decisions = _safe_int( + _deep_find_first( + summary, + ( + "valid_decisions", + "valid_records", + "decisions", + "records", + "target_reached_decisions", + ), + ) + ) + games = _safe_int( + _deep_find_first( + summary, + ( + "shadow_games", + "games", + "completed_games", + "games_completed", + "game_count", + ), + ) + ) + invalid_lines = _safe_int( + _deep_find_first( + summary, + ( + "invalid_lines", + "invalid_records", + "invalid", + ), + ) + ) + + ml_used = _safe_int( + _deep_find_first( + summary, + ( + "guarded_direct_ranker_moves", + "elite_direct_ranker_moves", + "ml_used", + "used_direct_ranker", + ), + ) + ) + fallback = _safe_int( + _deep_find_first( + summary, + ( + "guarded_fallback_moves", + "elite_fallback_moves", + "fallback", + "fallback_count", + ), + ) + ) + total_guarded = ml_used + fallback + fallback_percent = ( + round(fallback * 100.0 / total_guarded, 2) + if total_guarded > 0 + else 0.0 + ) + + move_policy = normalize_policy( + _deep_find_first(summary, ("move_policy", "policy")) + ) + phase_counts = _phase_counts(summary) + fallback_reasons = _fallback_reasons(summary) + + target_reached_value = _deep_find_first( + summary, + ("target_reached", "target", "target_met"), + ) + if isinstance(target_reached_value, bool): + target_reached = target_reached_value + else: + target_reached = decisions > 0 + + return { + "move_policy": move_policy, + "decisions": decisions, + "games": games, + "invalid_lines": invalid_lines, + "ml_used": ml_used, + "fallback": fallback, + "fallback_percent": fallback_percent, + "phase_counts": phase_counts, + "fallback_reasons": fallback_reasons, + "target_reached": target_reached, + } + + +def evaluate_elite_guarded_evidence( + *, + username: str, + time_class: str, + summary: Mapping[str, Any], + summary_path: Path | str, + output_root: Path | str = DEFAULT_OUTPUT_ROOT, + thresholds: EliteEvidenceThresholds | None = None, + expected_policy: str | None = None, +) -> EliteEvidenceResult: + thresholds = thresholds or EliteEvidenceThresholds() + metrics = extract_elite_selfplay_metrics(summary) + move_policy = normalize_policy(expected_policy or metrics.get("move_policy")) + + failed: list[str] = [] + notes: list[str] = [] + + if not is_elite_policy(move_policy): + failed.append("policy.not_elite_guarded") + notes.append( + "Self-play summary must use direct_ranker_elite_guarded or direct_ranker_elite_guarded_strict." + ) + + if metrics["decisions"] < thresholds.minimum_decisions: + failed.append("coverage.decisions") + if metrics["games"] < thresholds.minimum_games: + failed.append("coverage.games") + + for phase, count in metrics["phase_counts"].items(): + if int(count) < thresholds.minimum_phase_records: + failed.append(f"coverage.phase.{phase}") + + if thresholds.block_on_invalid_lines and metrics["invalid_lines"] > 0: + failed.append("integrity.invalid_lines") + + fallback_limit = ( + thresholds.strict_max_fallback_percent + if move_policy == DIRECT_RANKER_ELITE_GUARDED_STRICT + else thresholds.max_fallback_percent + ) + if metrics["fallback_percent"] > fallback_limit: + failed.append("runtime.fallback_percent") + + hard_error_reasons = { + "guarded_hybrid_error", + "direct_ranker_unavailable", + "shadow_unavailable", + "model_unavailable", + } + fallback_reasons = _as_mapping(metrics.get("fallback_reasons")) + for reason in hard_error_reasons: + if _safe_int(fallback_reasons.get(reason)) > 0: + failed.append(f"runtime.fallback_reason.{reason}") + + if any(item.startswith("integrity.") for item in failed) or any( + item.startswith("runtime.fallback_reason.") for item in failed + ): + decision = "elite_guarded_blocked" + reason = "runtime_or_integrity_blocker" + elif failed: + decision = "needs_more_evidence" + reason = "insufficient_elite_selfplay_evidence" + else: + decision = expected_ready_decision(move_policy) + reason = "elite_guarded_evidence_ready" + + if move_policy == DIRECT_RANKER_ELITE_GUARDED_STRICT: + notes.append( + "Strict policy allows higher fallback rate because stricter guardrails are expected to fallback more often." + ) + + return EliteEvidenceResult( + schema_version=ELITE_EVIDENCE_SCHEMA_VERSION, + generated_at=_utc_now_iso(), + username=username.strip().lower(), + time_class=time_class.strip().lower(), + move_policy=move_policy, + decision=decision, + reason=reason, + recommended_policy=recommended_policy_for_decision( + decision=decision, + move_policy=move_policy, + ), + selfplay_summary_path=str(Path(summary_path)), + output_root=str(Path(output_root)), + thresholds=asdict(thresholds), + metrics=dict(metrics), + failed_checks=tuple(failed), + notes=tuple(notes), + ) + + +def load_selfplay_summary(path: Path | str) -> dict[str, Any]: + with Path(path).open("r", encoding="utf-8") as file: + data = json.load(file) + if not isinstance(data, dict): + raise ValueError("Self-play summary JSON must contain an object.") + return data + + +def result_to_json_dict(result: EliteEvidenceResult) -> dict[str, Any]: + payload = asdict(result) + payload["failed_checks"] = list(result.failed_checks) + payload["notes"] = list(result.notes) + return payload + + +def render_elite_guarded_evidence_markdown( + result: EliteEvidenceResult, +) -> str: + metrics = result.metrics + phases = _as_mapping(metrics.get("phase_counts")) + fallback_reasons = _as_mapping(metrics.get("fallback_reasons")) + + lines = [ + "# ChessPersona Elite Guarded Evidence Gate", + "", + f"Generated: `{result.generated_at}`", + "", + "## Decision", + "", + f"- Username: `{result.username}`", + f"- Time class: `{result.time_class}`", + f"- Move policy: `{result.move_policy}`", + f"- Decision: `{result.decision}`", + f"- Reason: `{result.reason}`", + f"- Recommended policy: `{result.recommended_policy}`", + "", + "## Metrics", + "", + "| Metric | Value |", + "|---|---:|", + f"| Decisions | {metrics.get('decisions', 0)} |", + f"| Games | {metrics.get('games', 0)} |", + f"| Invalid lines | {metrics.get('invalid_lines', 0)} |", + f"| ML used | {metrics.get('ml_used', 0)} |", + f"| Fallback | {metrics.get('fallback', 0)} |", + f"| Fallback percent | {metrics.get('fallback_percent', 0.0)}% |", + "", + "## Phase coverage", + "", + "| Phase | Records |", + "|---|---:|", + ] + + for phase in ("opening", "middlegame", "endgame"): + lines.append(f"| {phase} | {phases.get(phase, 0)} |") + + lines.extend( + [ + "", + "## Fallback reasons", + "", + "| Reason | Count |", + "|---|---:|", + ] + ) + if fallback_reasons: + for reason, count in sorted(fallback_reasons.items()): + lines.append(f"| `{reason}` | {count} |") + else: + lines.append("| — | 0 |") + + lines.extend( + [ + "", + "## Failed checks", + "", + ] + ) + if result.failed_checks: + for check in result.failed_checks: + lines.append(f"- `{check}`") + else: + lines.append("- none") + + if result.notes: + lines.extend(["", "## Notes", ""]) + for note in result.notes: + lines.append(f"- {note}") + + lines.extend( + [ + "", + "## Meaning", + "", + "- This gate is only for elite model-first guarded runtime evidence.", + "- It does not change the normal Direct Ranker release gate.", + "- Normal guarded beta still requires `release_candidate_ready`.", + "- Elite guarded mode is allowed only when self-play evidence is sufficient and guardrails are healthy.", + "", + ] + ) + + return "\n".join(lines) + + +def write_elite_guarded_evidence_report( + result: EliteEvidenceResult, + *, + output_root: Path | str = DEFAULT_OUTPUT_ROOT, +) -> tuple[Path, Path]: + json_path, markdown_path = evidence_output_paths( + username=result.username, + time_class=result.time_class, + output_root=output_root, + ) + json_path.parent.mkdir(parents=True, exist_ok=True) + json_path.write_text( + json.dumps(result_to_json_dict(result), indent=2, ensure_ascii=False) + + "\n", + encoding="utf-8", + ) + markdown_path.write_text( + render_elite_guarded_evidence_markdown(result), + encoding="utf-8", + ) + return json_path, markdown_path + + +def build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Evaluate elite guarded self-play evidence.", + ) + parser.add_argument("--username", required=True) + parser.add_argument("--time-class", required=True, choices=("rapid", "blitz")) + parser.add_argument( + "--move-policy", + default="", + help=( + "Expected elite policy. Defaults to the policy stored in self-play summary." + ), + ) + parser.add_argument( + "--selfplay-root", + default=str(DEFAULT_SELFPLAY_ROOT), + ) + parser.add_argument( + "--summary", + default="", + help="Optional explicit selfplay_summary.json path.", + ) + parser.add_argument( + "--output-root", + default=str(DEFAULT_OUTPUT_ROOT), + ) + parser.add_argument("--minimum-decisions", type=int, default=320) + parser.add_argument("--minimum-games", type=int, default=8) + parser.add_argument("--minimum-phase-records", type=int, default=20) + parser.add_argument("--max-fallback-percent", type=float, default=50.0) + parser.add_argument( + "--strict-max-fallback-percent", + type=float, + default=70.0, + ) + parser.add_argument( + "--allow-invalid-lines", + action="store_true", + help="Do not block on invalid lines.", + ) + parser.add_argument( + "--strict", + action="store_true", + help="Exit 2 when decision is not elite ready.", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_arg_parser().parse_args(argv) + + summary_path = ( + Path(args.summary) + if args.summary + else selfplay_summary_path( + username=args.username, + time_class=args.time_class, + root=args.selfplay_root, + ) + ) + + thresholds = EliteEvidenceThresholds( + minimum_decisions=args.minimum_decisions, + minimum_games=args.minimum_games, + minimum_phase_records=args.minimum_phase_records, + max_fallback_percent=args.max_fallback_percent, + strict_max_fallback_percent=args.strict_max_fallback_percent, + block_on_invalid_lines=not args.allow_invalid_lines, + ) + + summary = load_selfplay_summary(summary_path) + result = evaluate_elite_guarded_evidence( + username=args.username, + time_class=args.time_class, + summary=summary, + summary_path=summary_path, + output_root=args.output_root, + thresholds=thresholds, + expected_policy=args.move_policy or None, + ) + json_path, markdown_path = write_elite_guarded_evidence_report( + result, + output_root=args.output_root, + ) + + print("ChessPersona Elite Guarded Evidence Gate") + print("=" * 48) + print(f"Username : {result.username}") + print(f"Time class : {result.time_class}") + print(f"Move policy : {result.move_policy}") + print(f"Decision : {result.decision}") + print(f"Reason : {result.reason}") + print(f"Recommended policy : {result.recommended_policy}") + print(f"Failed checks : {len(result.failed_checks)}") + for check in result.failed_checks: + print(f" - {check}") + print(f"JSON : {json_path}") + print(f"Markdown : {markdown_path}") + + ready = result.decision in { + "elite_guarded_ready", + "elite_guarded_strict_ready", + } + if args.strict and not ready: + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + +# Step 32E.1 hotfix: +# Prefer final_evidence over initial_evidence and reject stale non-elite summaries +# when an elite policy is expected. + +def _summary_policy(summary: Mapping[str, Any]) -> str: + config = _as_mapping(summary.get("config")) + policy = normalize_policy(config.get("move_policy")) + if policy: + return policy + return normalize_policy(summary.get("move_policy")) + + +def extract_elite_selfplay_metrics( + summary: Mapping[str, Any], +) -> dict[str, Any]: + final_evidence = _as_mapping(summary.get("final_evidence")) + guarded = _as_mapping(summary.get("guarded_hybrid")) + + game_results = _as_sequence(summary.get("game_results")) + game_ml_used = 0 + game_fallback = 0 + game_fallback_reasons: dict[str, int] = {} + for item in game_results: + if not isinstance(item, Mapping): + continue + game_ml_used += _safe_int(item.get("guarded_direct_ranker_moves")) + game_fallback += _safe_int(item.get("guarded_fallback_moves")) + reasons = _as_mapping(item.get("guarded_fallback_reasons")) + for reason, count in reasons.items(): + key = str(reason) + game_fallback_reasons[key] = ( + game_fallback_reasons.get(key, 0) + + _safe_int(count) + ) + + phase_payload = _as_mapping(final_evidence.get("phase_counts")) + decisions = _safe_int( + final_evidence.get("valid_records") + or final_evidence.get("valid_decisions") + or final_evidence.get("records") + ) + games = _safe_int( + final_evidence.get("shadow_games") + or final_evidence.get("games") + or summary.get("games_completed") + or summary.get("games_started") + ) + invalid_lines = _safe_int(final_evidence.get("invalid_lines")) + + ml_used = _safe_int( + guarded.get("direct_ranker_moves") + or guarded.get("ml_used") + or game_ml_used + ) + fallback = _safe_int( + guarded.get("fallback_moves") + or guarded.get("fallback") + or game_fallback + ) + total_guarded = ml_used + fallback + fallback_percent = ( + round(fallback * 100.0 / total_guarded, 2) + if total_guarded > 0 + else 0.0 + ) + + fallback_reasons = { + str(key): _safe_int(value) + for key, value in _as_mapping( + guarded.get("fallback_reasons") + ).items() + } + if not fallback_reasons and game_fallback_reasons: + fallback_reasons = game_fallback_reasons + + target_reached_value = summary.get("target_reached") + if isinstance(target_reached_value, bool): + target_reached = target_reached_value + else: + target_reached = decisions > 0 + + return { + "move_policy": _summary_policy(summary), + "decisions": decisions, + "games": games, + "invalid_lines": invalid_lines, + "ml_used": ml_used, + "fallback": fallback, + "fallback_percent": fallback_percent, + "phase_counts": { + "opening": _safe_int(phase_payload.get("opening")), + "middlegame": _safe_int(phase_payload.get("middlegame")), + "endgame": _safe_int(phase_payload.get("endgame")), + }, + "fallback_reasons": fallback_reasons, + "target_reached": target_reached, + } + + +def evaluate_elite_guarded_evidence( + *, + username: str, + time_class: str, + summary: Mapping[str, Any], + summary_path: Path | str, + output_root: Path | str = DEFAULT_OUTPUT_ROOT, + thresholds: EliteEvidenceThresholds | None = None, + expected_policy: str | None = None, +) -> EliteEvidenceResult: + thresholds = thresholds or EliteEvidenceThresholds() + metrics = extract_elite_selfplay_metrics(summary) + summary_policy = normalize_policy(metrics.get("move_policy")) + move_policy = normalize_policy(expected_policy or summary_policy) + + failed: list[str] = [] + notes: list[str] = [] + + if not is_elite_policy(move_policy): + failed.append("policy.not_elite_guarded") + notes.append( + "Self-play summary must use direct_ranker_elite_guarded or direct_ranker_elite_guarded_strict." + ) + + if ( + expected_policy + and summary_policy + and summary_policy != move_policy + ): + failed.append("policy.summary_mismatch") + notes.append( + f"Expected {move_policy}, but self-play summary was generated with {summary_policy}. " + "Run self-play again with --reset-logs for the elite policy." + ) + + if metrics["decisions"] < thresholds.minimum_decisions: + failed.append("coverage.decisions") + if metrics["games"] < thresholds.minimum_games: + failed.append("coverage.games") + + for phase, count in metrics["phase_counts"].items(): + if int(count) < thresholds.minimum_phase_records: + failed.append(f"coverage.phase.{phase}") + + if thresholds.block_on_invalid_lines and metrics["invalid_lines"] > 0: + failed.append("integrity.invalid_lines") + + fallback_limit = ( + thresholds.strict_max_fallback_percent + if move_policy == DIRECT_RANKER_ELITE_GUARDED_STRICT + else thresholds.max_fallback_percent + ) + if metrics["fallback_percent"] > fallback_limit: + failed.append("runtime.fallback_percent") + + hard_error_reasons = { + "guarded_hybrid_error", + "direct_ranker_unavailable", + "shadow_unavailable", + "model_unavailable", + } + fallback_reasons = _as_mapping(metrics.get("fallback_reasons")) + for reason in hard_error_reasons: + if _safe_int(fallback_reasons.get(reason)) > 0: + failed.append(f"runtime.fallback_reason.{reason}") + + if any(item.startswith("integrity.") for item in failed) or any( + item.startswith("runtime.fallback_reason.") for item in failed + ): + decision = "elite_guarded_blocked" + reason = "runtime_or_integrity_blocker" + elif failed: + decision = "needs_more_evidence" + reason = "insufficient_elite_selfplay_evidence" + else: + decision = expected_ready_decision(move_policy) + reason = "elite_guarded_evidence_ready" + + if move_policy == DIRECT_RANKER_ELITE_GUARDED_STRICT: + notes.append( + "Strict policy allows higher fallback rate because stricter guardrails are expected to fallback more often." + ) + + return EliteEvidenceResult( + schema_version=ELITE_EVIDENCE_SCHEMA_VERSION, + generated_at=_utc_now_iso(), + username=username.strip().lower(), + time_class=time_class.strip().lower(), + move_policy=move_policy, + decision=decision, + reason=reason, + recommended_policy=recommended_policy_for_decision( + decision=decision, + move_policy=move_policy, + ), + selfplay_summary_path=str(Path(summary_path)), + output_root=str(Path(output_root)), + thresholds=asdict(thresholds), + metrics=dict(metrics), + failed_checks=tuple(failed), + notes=tuple(notes), + ) + +# Step 32E.2 hotfix: +# Keep compatibility with synthetic/unit summaries that store metrics at the +# top level, while still preferring real self-play final_evidence. + +def extract_elite_selfplay_metrics( + summary: Mapping[str, Any], +) -> dict[str, Any]: + final_evidence = _as_mapping(summary.get("final_evidence")) + evidence = final_evidence if final_evidence else summary + guarded = _as_mapping(summary.get("guarded_hybrid")) + + game_results = _as_sequence(summary.get("game_results")) + game_ml_used = 0 + game_fallback = 0 + game_fallback_reasons: dict[str, int] = {} + for item in game_results: + if not isinstance(item, Mapping): + continue + game_ml_used += _safe_int(item.get("guarded_direct_ranker_moves")) + game_fallback += _safe_int(item.get("guarded_fallback_moves")) + reasons = _as_mapping(item.get("guarded_fallback_reasons")) + for reason, count in reasons.items(): + key = str(reason) + game_fallback_reasons[key] = ( + game_fallback_reasons.get(key, 0) + + _safe_int(count) + ) + + phase_payload = _as_mapping(evidence.get("phase_counts")) + decisions = _safe_int( + evidence.get("valid_records") + or evidence.get("valid_decisions") + or evidence.get("decisions") + or evidence.get("records") + or summary.get("valid_records") + or summary.get("valid_decisions") + or summary.get("decisions") + or summary.get("records") + ) + games = _safe_int( + evidence.get("shadow_games") + or evidence.get("games") + or evidence.get("completed_games") + or evidence.get("games_completed") + or summary.get("shadow_games") + or summary.get("games") + or summary.get("completed_games") + or summary.get("games_completed") + or summary.get("games_started") + or summary.get("game_count") + ) + invalid_lines = _safe_int( + evidence.get("invalid_lines") + or evidence.get("invalid_records") + or summary.get("invalid_lines") + or summary.get("invalid_records") + ) + + ml_used = _safe_int( + guarded.get("direct_ranker_moves") + or guarded.get("ml_used") + or guarded.get("guarded_direct_ranker_moves") + or summary.get("guarded_direct_ranker_moves") + or summary.get("elite_direct_ranker_moves") + or summary.get("ml_used") + or summary.get("used_direct_ranker") + or game_ml_used + ) + fallback = _safe_int( + guarded.get("fallback_moves") + or guarded.get("fallback") + or guarded.get("guarded_fallback_moves") + or summary.get("guarded_fallback_moves") + or summary.get("elite_fallback_moves") + or summary.get("fallback") + or summary.get("fallback_count") + or game_fallback + ) + total_guarded = ml_used + fallback + fallback_percent = ( + round(fallback * 100.0 / total_guarded, 2) + if total_guarded > 0 + else 0.0 + ) + + fallback_reasons = { + str(key): _safe_int(value) + for key, value in _as_mapping( + guarded.get("fallback_reasons") + ).items() + } + if not fallback_reasons: + fallback_reasons = { + str(key): _safe_int(value) + for key, value in _as_mapping( + summary.get("guarded_fallback_reasons") + or summary.get("fallback_reasons") + or summary.get("elite_fallback_reasons") + ).items() + } + if not fallback_reasons and game_fallback_reasons: + fallback_reasons = game_fallback_reasons + + target_reached_value = summary.get("target_reached") + if isinstance(target_reached_value, bool): + target_reached = target_reached_value + else: + target_reached = decisions > 0 + + return { + "move_policy": _summary_policy(summary), + "decisions": decisions, + "games": games, + "invalid_lines": invalid_lines, + "ml_used": ml_used, + "fallback": fallback, + "fallback_percent": fallback_percent, + "phase_counts": { + "opening": _safe_int(phase_payload.get("opening")), + "middlegame": _safe_int(phase_payload.get("middlegame")), + "endgame": _safe_int(phase_payload.get("endgame")), + }, + "fallback_reasons": fallback_reasons, + "target_reached": target_reached, + } diff --git a/src/ml/direct_ranker_guarded_hybrid.py b/src/ml/direct_ranker_guarded_hybrid.py index f569b1f..67e1a33 100644 --- a/src/ml/direct_ranker_guarded_hybrid.py +++ b/src/ml/direct_ranker_guarded_hybrid.py @@ -789,3 +789,36 @@ def render_guarded_hybrid_markdown(report: Mapping[str, Any]) -> str: ] ) return "\n".join(lines) + +# Step 32E.3 hotfix: +# Elite synthetic release reports created by UI/self-play use a top-level +# decision field. Normal release-gate reports store status under assessment. +# Support both shapes so elite guarded runtime can bypass the normal blocked +# release gate only when its synthetic report explicitly says ready. + +_previous_release_gate_status_step32e3 = release_gate_status + + +def release_gate_status(report: Mapping[str, Any] | None) -> str: + status = _previous_release_gate_status_step32e3(report) + if status: + return status + + if not isinstance(report, Mapping): + return "" + + for key in ("decision", "status", "release_gate_status"): + value = report.get(key) + text = str(value or "").strip() + if text: + return text + + assessment = report.get("assessment", {}) + if isinstance(assessment, Mapping): + for key in ("decision", "status"): + value = assessment.get(key) + text = str(value or "").strip() + if text: + return text + + return "" diff --git a/src/selfplay/direct_ranker_selfplay.py b/src/selfplay/direct_ranker_selfplay.py index 6c56807..c60d7b6 100644 --- a/src/selfplay/direct_ranker_selfplay.py +++ b/src/selfplay/direct_ranker_selfplay.py @@ -1036,26 +1036,20 @@ def run_single_game( guarded_hybrid: DirectRankerGuardedHybrid | None = None if config.move_policy in MOVE_POLICY_GUARDED_RUNTIME_POLICIES: - release_report = read_release_gate_report( - release_gate_path( - config.username, - config.time_class, - config.release_gate_root, - ) + release_report = release_gate_report_for_move_policy( + move_policy=config.move_policy, + username=config.username, + time_class=config.time_class, + release_gate_root=config.release_gate_root, ) guarded_hybrid = DirectRankerGuardedHybrid( shadow=shadow, release_gate_report=release_report, - config=GuardedHybridConfig( - max_engine_rank=( - config.guarded_max_engine_rank - ), - max_score_gap_cp=( - config.guarded_max_score_gap_cp - ), - min_probability=( - config.guarded_min_probability - ), + config=guarded_config_for_move_policy( + move_policy=config.move_policy, + max_engine_rank=config.guarded_max_engine_rank, + max_score_gap_cp=config.guarded_max_score_gap_cp, + min_probability=config.guarded_min_probability, ), ) @@ -1554,3 +1548,80 @@ def submit_one() -> bool: summary_path ) return summary + +# Step 32E.4 hotfix: +# Elite model-first guarded self-play must not be blocked by the normal guarded +# beta release gate. Normal direct_ranker_guarded still requires +# release_candidate_ready. Elite evidence is judged by the separate +# elite guarded evidence gate. + +_previous_guarded_config_for_move_policy_step32e4 = guarded_config_for_move_policy +_previous_release_gate_report_for_move_policy_step32e4 = release_gate_report_for_move_policy + + +def release_gate_report_for_move_policy( + *, + move_policy: str, + username: str, + time_class: str, + release_gate_root: Path, +) -> dict[str, object]: + policy = str(move_policy or "").strip().lower() + + if is_elite_guarded_policy(policy): + return { + "decision": "release_candidate_ready", + "status": "release_candidate_ready", + "assessment": { + "status": "release_candidate_ready", + "decision": "release_candidate_ready", + }, + "profile": "elite_model_first_guarded_selfplay_v2", + "elite_engine_like_persona": True, + "elite_guarded_policy": policy, + "username": username, + "time_class": time_class, + "note": ( + "Elite model-first guarded self-play bypasses the normal " + "guarded beta release gate. The separate elite evidence gate " + "must still approve runtime evidence before the policy is " + "considered ready." + ), + } + + return _previous_release_gate_report_for_move_policy_step32e4( + move_policy=move_policy, + username=username, + time_class=time_class, + release_gate_root=release_gate_root, + ) + + +def guarded_config_for_move_policy( + *, + move_policy: str, + max_engine_rank: int = 5, + max_score_gap_cp: float = 75.0, + min_probability: float = 0.0, +) -> GuardedHybridConfig: + policy = str(move_policy or "").strip().lower() + config = _previous_guarded_config_for_move_policy_step32e4( + move_policy=policy, + max_engine_rank=max_engine_rank, + max_score_gap_cp=max_score_gap_cp, + min_probability=min_probability, + ) + + if is_elite_guarded_policy(policy): + return GuardedHybridConfig( + require_release_gate_ready=False, + required_release_status=config.required_release_status, + expected_model_schema_version=config.expected_model_schema_version, + expected_feature_schema_version=config.expected_feature_schema_version, + expected_feature_profile=config.expected_feature_profile, + max_engine_rank=config.max_engine_rank, + max_score_gap_cp=config.max_score_gap_cp, + min_probability=config.min_probability, + ).validated() + + return config diff --git a/src/ui/workers.py b/src/ui/workers.py index 7032828..e7e2a8d 100644 --- a/src/ui/workers.py +++ b/src/ui/workers.py @@ -362,3 +362,172 @@ def run(self) -> None: self.signals.error.emit( f"{type(exc).__name__}: {exc}" ) + +# Step 32E.4 hotfix: +# Keep UI runtime consistent with headless self-play: elite guarded policies are +# judged by elite evidence, not by the normal guarded beta release gate. + +_previous_elite_release_gate_report_step32e4 = _elite_release_gate_report +_previous_elite_guarded_config_step32e4 = _elite_guarded_config + + +def _elite_release_gate_report( + *, + policy: str, + username: str, + time_class: str, +) -> dict[str, Any]: + report = _previous_elite_release_gate_report_step32e4( + policy=policy, + username=username, + time_class=time_class, + ) + report["decision"] = "release_candidate_ready" + report["status"] = "release_candidate_ready" + report["assessment"] = { + "status": "release_candidate_ready", + "decision": "release_candidate_ready", + } + report["profile"] = "elite_model_first_guarded_runtime_v2" + return report + + +def _elite_guarded_config(policy: str) -> GuardedHybridConfig: + config = _previous_elite_guarded_config_step32e4(policy) + return GuardedHybridConfig( + require_release_gate_ready=False, + required_release_status=config.required_release_status, + expected_model_schema_version=config.expected_model_schema_version, + expected_feature_schema_version=config.expected_feature_schema_version, + expected_feature_profile=config.expected_feature_profile, + max_engine_rank=config.max_engine_rank, + max_score_gap_cp=config.max_score_gap_cp, + min_probability=config.min_probability, + ).validated() + +# Step 32E.5 hotfix: +# The Step 32E.4 UI helper copied optional GuardedHybridConfig attributes +# too aggressively. Some project versions do not expose every optional +# attribute, which makes the worker catch the exception and fall back to the +# statistical selector. Build the elite config with only stable constructor +# arguments so the elite guarded branch remains model-first. + +def _elite_guarded_config(policy: str) -> GuardedHybridConfig: + guardrails = guardrails_for_policy(policy) + + return GuardedHybridConfig( + require_release_gate_ready=False, + max_engine_rank=int( + _guardrail_number( + guardrails, + ( + "max_engine_rank", + "max_candidate_rank", + "max_rank", + ), + 5, + ) + ), + max_score_gap_cp=_guardrail_number( + guardrails, + ( + "max_score_gap_cp", + "max_eval_gap_cp", + "max_gap_cp", + "max_centipawn_gap_cp", + ), + 75.0, + ), + min_probability=_guardrail_number( + guardrails, + ( + "min_probability", + "minimum_probability", + "min_model_probability", + ), + 0.0, + ), + ).validated() + +# Step 32E.6 hotfix: +# choose_bot_turn_payload calls _elite_release_gate_report(move_policy=...) +# and _elite_guarded_config(policy, guarded_config). Earlier appended hotfixes +# narrowed these helper signatures, causing a TypeError that was caught by the +# worker and silently fell back to the statistical selector. Keep the helpers +# compatible with the original call site. + +def _elite_release_gate_report( + *, + move_policy: str = "", + policy: str = "", + username: str, + time_class: str, +) -> dict[str, Any]: + resolved_policy = str(move_policy or policy or "").strip().lower() + return { + "decision": "release_candidate_ready", + "status": "release_candidate_ready", + "assessment": { + "status": "release_candidate_ready", + "decision": "release_candidate_ready", + }, + "profile": "elite_model_first_guarded_runtime_v3", + "elite_engine_like_persona": True, + "elite_guarded_policy": resolved_policy, + "username": username, + "time_class": time_class, + "note": ( + "Elite model-first guarded runtime bypasses the normal guarded " + "beta release gate. Runtime readiness is judged by the separate " + "elite guarded evidence gate." + ), + } + + +def _elite_guarded_config( + move_policy: str, + guarded_config: GuardedHybridConfig | None = None, +) -> GuardedHybridConfig: + if guarded_config is not None: + return GuardedHybridConfig( + require_release_gate_ready=False, + max_engine_rank=guarded_config.max_engine_rank, + max_score_gap_cp=guarded_config.max_score_gap_cp, + min_probability=guarded_config.min_probability, + ).validated() + + guardrails = guardrails_for_policy(move_policy) + + return GuardedHybridConfig( + require_release_gate_ready=False, + max_engine_rank=int( + _guardrail_number( + guardrails, + ( + "max_engine_rank", + "max_candidate_rank", + "max_rank", + ), + 5, + ) + ), + max_score_gap_cp=_guardrail_number( + guardrails, + ( + "max_score_gap_cp", + "max_eval_gap_cp", + "max_gap_cp", + "max_centipawn_gap_cp", + ), + 75.0, + ), + min_probability=_guardrail_number( + guardrails, + ( + "min_probability", + "minimum_probability", + "min_model_probability", + ), + 0.0, + ), + ).validated() diff --git a/tests/test_elite_guarded_evidence_gate.py b/tests/test_elite_guarded_evidence_gate.py new file mode 100644 index 0000000..eb17755 --- /dev/null +++ b/tests/test_elite_guarded_evidence_gate.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from src.ml.direct_ranker_elite_guarded_evidence import ( + DIRECT_RANKER_ELITE_GUARDED, + DIRECT_RANKER_ELITE_GUARDED_STRICT, + EliteEvidenceThresholds, + evaluate_elite_guarded_evidence, + extract_elite_selfplay_metrics, + main, +) + + +def _summary( + *, + policy: str = DIRECT_RANKER_ELITE_GUARDED, + decisions: int = 340, + games: int = 9, + opening: int = 90, + middlegame: int = 200, + endgame: int = 50, + invalid_lines: int = 0, + ml_used: int = 260, + fallback: int = 60, +) -> dict[str, object]: + return { + "move_policy": policy, + "valid_decisions": decisions, + "shadow_games": games, + "invalid_lines": invalid_lines, + "guarded_direct_ranker_moves": ml_used, + "guarded_fallback_moves": fallback, + "guarded_fallback_reasons": {}, + "phase_counts": { + "opening": opening, + "middlegame": middlegame, + "endgame": endgame, + }, + } + + +def test_extract_elite_selfplay_metrics_from_summary() -> None: + metrics = extract_elite_selfplay_metrics(_summary()) + + assert metrics["move_policy"] == DIRECT_RANKER_ELITE_GUARDED + assert metrics["decisions"] == 340 + assert metrics["games"] == 9 + assert metrics["invalid_lines"] == 0 + assert metrics["ml_used"] == 260 + assert metrics["fallback"] == 60 + assert metrics["fallback_percent"] == 18.75 + assert metrics["phase_counts"]["endgame"] == 50 + + +def test_elite_guarded_ready_decision() -> None: + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="rapid", + summary=_summary(), + summary_path=Path("selfplay_summary.json"), + ) + + assert result.decision == "elite_guarded_ready" + assert result.reason == "elite_guarded_evidence_ready" + assert result.failed_checks == () + assert result.recommended_policy == "Direct Ranker Elite Guarded" + + +def test_elite_guarded_strict_ready_decision() -> None: + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="blitz", + summary=_summary( + policy=DIRECT_RANKER_ELITE_GUARDED_STRICT, + ml_used=190, + fallback=120, + ), + summary_path=Path("selfplay_summary.json"), + ) + + assert result.decision == "elite_guarded_strict_ready" + assert result.recommended_policy == "Direct Ranker Elite Guarded Strict" + + +def test_elite_guarded_needs_more_evidence() -> None: + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="rapid", + summary=_summary(decisions=80, games=2, endgame=4), + summary_path=Path("selfplay_summary.json"), + ) + + assert result.decision == "needs_more_evidence" + assert "coverage.decisions" in result.failed_checks + assert "coverage.games" in result.failed_checks + assert "coverage.phase.endgame" in result.failed_checks + + +def test_elite_guarded_blocks_invalid_lines() -> None: + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="blitz", + summary=_summary(invalid_lines=1), + summary_path=Path("selfplay_summary.json"), + ) + + assert result.decision == "elite_guarded_blocked" + assert "integrity.invalid_lines" in result.failed_checks + + +def test_elite_guarded_blocks_hard_fallback_reason() -> None: + summary = _summary() + summary["guarded_fallback_reasons"] = {"guarded_hybrid_error": 1} + + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="blitz", + summary=summary, + summary_path=Path("selfplay_summary.json"), + ) + + assert result.decision == "elite_guarded_blocked" + assert "runtime.fallback_reason.guarded_hybrid_error" in result.failed_checks + + +def test_elite_guarded_cli_writes_reports(tmp_path: Path) -> None: + summary_path = tmp_path / "selfplay_summary.json" + summary_path.write_text( + json.dumps(_summary()), + encoding="utf-8", + ) + output_root = tmp_path / "out" + + code = main( + [ + "--username", + "hikaru", + "--time-class", + "rapid", + "--summary", + str(summary_path), + "--output-root", + str(output_root), + "--strict", + ] + ) + + assert code == 0 + assert ( + output_root + / "hikaru" + / "rapid_elite_guarded_evidence.json" + ).exists() + assert ( + output_root + / "hikaru" + / "rapid_elite_guarded_evidence.md" + ).exists() + + +def test_elite_guarded_cli_strict_returns_two_when_not_ready( + tmp_path: Path, +) -> None: + summary_path = tmp_path / "selfplay_summary.json" + summary_path.write_text( + json.dumps(_summary(decisions=10, games=1)), + encoding="utf-8", + ) + + code = main( + [ + "--username", + "hikaru", + "--time-class", + "rapid", + "--summary", + str(summary_path), + "--output-root", + str(tmp_path / "out"), + "--strict", + ] + ) + + assert code == 2 + +def test_extract_metrics_prefers_final_evidence_over_initial_evidence() -> None: + summary = { + "target_reached": True, + "initial_evidence": { + "valid_records": 0, + "shadow_games": 0, + "invalid_lines": 0, + "phase_counts": { + "opening": 0, + "middlegame": 0, + "endgame": 0, + }, + }, + "final_evidence": { + "valid_records": 413, + "shadow_games": 12, + "invalid_lines": 0, + "phase_counts": { + "opening": 120, + "middlegame": 267, + "endgame": 26, + }, + }, + "config": { + "move_policy": DIRECT_RANKER_ELITE_GUARDED, + }, + "guarded_hybrid": { + "direct_ranker_moves": 300, + "fallback_moves": 20, + "fallback_reasons": {}, + }, + } + + metrics = extract_elite_selfplay_metrics(summary) + + assert metrics["decisions"] == 413 + assert metrics["games"] == 12 + assert metrics["phase_counts"]["opening"] == 120 + assert metrics["phase_counts"]["middlegame"] == 267 + assert metrics["phase_counts"]["endgame"] == 26 + + +def test_expected_elite_policy_rejects_stale_shadow_summary() -> None: + summary = { + "target_reached": True, + "final_evidence": { + "valid_records": 413, + "shadow_games": 12, + "invalid_lines": 0, + "phase_counts": { + "opening": 120, + "middlegame": 267, + "endgame": 26, + }, + }, + "config": { + "move_policy": "direct_ranker_shadow", + }, + } + + result = evaluate_elite_guarded_evidence( + username="hikaru", + time_class="rapid", + summary=summary, + summary_path=Path("selfplay_summary.json"), + expected_policy=DIRECT_RANKER_ELITE_GUARDED, + ) + + assert result.decision == "needs_more_evidence" + assert "policy.summary_mismatch" in result.failed_checks + +def test_extract_metrics_supports_top_level_summary_shape() -> None: + metrics = extract_elite_selfplay_metrics( + { + "move_policy": DIRECT_RANKER_ELITE_GUARDED, + "valid_decisions": 88, + "shadow_games": 3, + "invalid_lines": 0, + "guarded_direct_ranker_moves": 70, + "guarded_fallback_moves": 10, + "guarded_fallback_reasons": {"engine_rank": 10}, + "phase_counts": { + "opening": 20, + "middlegame": 40, + "endgame": 28, + }, + } + ) + + assert metrics["decisions"] == 88 + assert metrics["games"] == 3 + assert metrics["ml_used"] == 70 + assert metrics["fallback"] == 10 + assert metrics["fallback_reasons"]["engine_rank"] == 10 + assert metrics["phase_counts"]["endgame"] == 28 diff --git a/tests/test_elite_guarded_runtime_wiring.py b/tests/test_elite_guarded_runtime_wiring.py index b8000b7..d0ea00f 100644 --- a/tests/test_elite_guarded_runtime_wiring.py +++ b/tests/test_elite_guarded_runtime_wiring.py @@ -202,3 +202,37 @@ def test_beta_dashboard_renders_elite_strict_policy() -> None: assert state.ml_used == 1 rendered = render_beta_dashboard(state) assert "Elite guarded strict" in rendered + +def test_ui_elite_guarded_config_bypasses_normal_release_gate() -> None: + from src.ui.workers import _elite_guarded_config + + config = _elite_guarded_config("direct_ranker_elite_guarded") + + assert config.require_release_gate_ready is False + +def test_ui_elite_guarded_config_can_construct_without_fallback() -> None: + from src.ui.workers import _elite_guarded_config + + config = _elite_guarded_config("direct_ranker_elite_guarded") + + assert config.require_release_gate_ready is False + assert config.max_engine_rank >= 1 + +def test_ui_elite_helpers_accept_original_worker_call_signature() -> None: + from src.ml.direct_ranker_guarded_hybrid import GuardedHybridConfig + from src.ui.workers import _elite_guarded_config, _elite_release_gate_report + + report = _elite_release_gate_report( + move_policy="direct_ranker_elite_guarded", + username="hikaru", + time_class="rapid", + ) + config = _elite_guarded_config( + "direct_ranker_elite_guarded", + GuardedHybridConfig(max_engine_rank=3), + ) + + assert report["decision"] == "release_candidate_ready" + assert report["assessment"]["status"] == "release_candidate_ready" + assert config.require_release_gate_ready is False + assert config.max_engine_rank == 3 diff --git a/tests/test_elite_guarded_selfplay_policy.py b/tests/test_elite_guarded_selfplay_policy.py index 50bce5f..42aed24 100644 --- a/tests/test_elite_guarded_selfplay_policy.py +++ b/tests/test_elite_guarded_selfplay_policy.py @@ -72,3 +72,49 @@ def test_elite_selfplay_uses_synthetic_release_report() -> None: assert report["decision"] == "release_candidate_ready" assert report["elite_engine_like_persona"] is True assert report["elite_guarded_policy"] == MOVE_POLICY_ELITE_GUARDED + +def test_elite_synthetic_release_report_is_ready() -> None: + from src.ml.direct_ranker_guarded_hybrid import ( + GuardedHybridConfig, + release_gate_is_ready, + release_gate_status, + ) + + report = { + "decision": "release_candidate_ready", + "elite_engine_like_persona": True, + "elite_guarded_policy": "direct_ranker_elite_guarded", + } + + assert release_gate_status(report) == "release_candidate_ready" + assert release_gate_is_ready(report, GuardedHybridConfig()) is True + +def test_elite_guarded_config_bypasses_normal_release_gate() -> None: + config = guarded_config_for_move_policy( + move_policy=MOVE_POLICY_ELITE_GUARDED, + ) + + assert config.require_release_gate_ready is False + + +def test_elite_selfplay_release_report_has_legacy_ready_shape() -> None: + report = release_gate_report_for_move_policy( + move_policy=MOVE_POLICY_ELITE_GUARDED, + username="hikaru", + time_class="rapid", + release_gate_root=Path("missing-release-gate-root"), + ) + + assert report["decision"] == "release_candidate_ready" + assert report["status"] == "release_candidate_ready" + assert report["assessment"]["status"] == "release_candidate_ready" + +def test_selfplay_worker_source_uses_elite_helper_initialization() -> None: + import inspect + import src.selfplay.direct_ranker_selfplay as selfplay + + source = inspect.getsource(selfplay) + + assert "release_gate_report_for_move_policy(" in source + assert "guarded_config_for_move_policy(" in source + assert "release_gate_report=release_report" in source