Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions check_runtime_readiness.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
from __future__ import annotations

from src.runtime.readiness import main


if __name__ == "__main__":
raise SystemExit(main())
1 change: 1 addition & 0 deletions src/runtime/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Runtime readiness helpers."""
364 changes: 364 additions & 0 deletions src/runtime/readiness.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,364 @@
from __future__ import annotations

import argparse
from dataclasses import dataclass
from pathlib import Path
from typing import Sequence

from src.engine.stockfish_service import DEFAULT_ENGINE_PATH
from src.ui.main_ui_elite_recommendation import (
MainUIEliteRecommendation,
build_main_ui_elite_recommendation,
)
from src.ui.move_policy import BOT_MOVE_POLICY_STATISTICAL

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")


@dataclass(frozen=True)
class RuntimeReadinessCheck:
name: str
ok: bool
status: str
path: Path | None = None
required: bool = True
detail: str = ""


@dataclass(frozen=True)
class RuntimeReadinessReport:
username: str
time_class: str
ready: bool
checks: tuple[RuntimeReadinessCheck, ...]
recommendation: MainUIEliteRecommendation

@property
def exit_code(self) -> int:
return 0 if self.ready else 2


def normalize_runtime_name(value: object) -> str:
return str(value or "").strip().lower()


def persona_candidate_paths(
*,
username: object,
time_class: object,
persona_root: Path | str = DEFAULT_PERSONA_ROOT,
) -> tuple[Path, ...]:
user = normalize_runtime_name(username)
tc = normalize_runtime_name(time_class)
root = Path(persona_root)
return (
root / user / f"{tc}_complete_persona.json",
root / user / f"{tc}_persona.json",
root / user / f"{tc}_description.json",
root / f"{user}_{tc}_complete_persona.json",
root / f"{user}_{tc}_persona.json",
)


def first_existing_path(paths: Sequence[Path]) -> Path | None:
for path in paths:
if path.exists():
return path
return None


def direct_ranker_model_path(
*,
username: object,
time_class: object,
models_root: Path | str = DEFAULT_DIRECT_MODELS_ROOT,
) -> Path:
user = normalize_runtime_name(username)
tc = normalize_runtime_name(time_class)
return Path(models_root) / user / f"{tc}_direct_ranker.joblib"


def direct_ranker_metrics_candidate_paths(
*,
username: object,
time_class: object,
models_root: Path | str = DEFAULT_DIRECT_MODELS_ROOT,
) -> tuple[Path, ...]:
user = normalize_runtime_name(username)
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,
path: Path,
required: bool = True,
ok_status: str = "OK",
missing_status: str = "MISSING",
) -> RuntimeReadinessCheck:
exists = path.exists()
return RuntimeReadinessCheck(
name=name,
ok=exists,
status=ok_status if exists else missing_status,
path=path,
required=required,
)


def _stockfish_check(engine_path: Path | str) -> RuntimeReadinessCheck:
return _path_check(
name="Stockfish",
path=Path(engine_path),
required=True,
)


def _persona_check(
*,
username: object,
time_class: object,
persona_root: Path | str,
) -> RuntimeReadinessCheck:
candidates = persona_candidate_paths(
username=username,
time_class=time_class,
persona_root=persona_root,
)
found = first_existing_path(candidates)
if found is not None:
return RuntimeReadinessCheck(
name="Persona",
ok=True,
status="OK",
path=found,
required=True,
)
return RuntimeReadinessCheck(
name="Persona",
ok=False,
status="MISSING",
path=candidates[0],
required=True,
detail="Tidak menemukan complete/persona/description JSON untuk username dan time class ini.",
)


def _model_check(
*,
username: object,
time_class: object,
models_root: Path | str,
) -> RuntimeReadinessCheck:
return _path_check(
name="Direct Ranker model",
path=direct_ranker_model_path(
username=username,
time_class=time_class,
models_root=models_root,
),
required=True,
)


def _metrics_check(
*,
username: object,
time_class: object,
models_root: Path | str,
strict_metrics: bool,
) -> RuntimeReadinessCheck:
candidates = direct_ranker_metrics_candidate_paths(
username=username,
time_class=time_class,
models_root=models_root,
)
found = first_existing_path(candidates)
if found is not None:
return RuntimeReadinessCheck(
name="Direct Ranker metrics",
ok=True,
status="OK",
path=found,
required=strict_metrics,
)
return RuntimeReadinessCheck(
name="Direct Ranker metrics",
ok=False,
status="MISSING",
path=candidates[0],
required=strict_metrics,
detail=(
"Metrics JSON tidak ditemukan. Secara default ini warning; "
"pakai --strict-metrics untuk menjadikannya blocking."
),
)


def _elite_evidence_check(
recommendation: MainUIEliteRecommendation,
) -> RuntimeReadinessCheck:
return RuntimeReadinessCheck(
name="Elite evidence",
ok=bool(recommendation.should_select_policy),
status=recommendation.decision,
path=recommendation.evidence_path,
required=True,
detail=recommendation.reason,
)


def _ui_runtime_check(
recommendation: MainUIEliteRecommendation,
) -> RuntimeReadinessCheck:
if recommendation.should_select_policy:
return RuntimeReadinessCheck(
name="UI runtime",
ok=True,
status="ready",
path=None,
required=True,
detail=recommendation.move_policy,
)
return RuntimeReadinessCheck(
name="UI runtime",
ok=False,
status="statistical_only",
path=None,
required=True,
detail="UI tidak boleh auto-select elite policy sampai evidence ready.",
)


def build_runtime_readiness_report(
*,
username: object,
time_class: object,
engine_path: Path | str = DEFAULT_ENGINE_PATH,
persona_root: Path | str = DEFAULT_PERSONA_ROOT,
models_root: Path | str = DEFAULT_DIRECT_MODELS_ROOT,
evidence_root: Path | str = DEFAULT_ELITE_EVIDENCE_ROOT,
strict_metrics: bool = False,
) -> RuntimeReadinessReport:
user = normalize_runtime_name(username)
tc = normalize_runtime_name(time_class)
recommendation = build_main_ui_elite_recommendation(
username=user,
time_class=tc,
evidence_root=evidence_root,
)
checks = (
_persona_check(
username=user,
time_class=tc,
persona_root=persona_root,
),
_stockfish_check(engine_path),
_model_check(
username=user,
time_class=tc,
models_root=models_root,
),
_metrics_check(
username=user,
time_class=tc,
models_root=models_root,
strict_metrics=strict_metrics,
),
_elite_evidence_check(recommendation),
_ui_runtime_check(recommendation),
)
ready = all(check.ok for check in checks if check.required)
return RuntimeReadinessReport(
username=user,
time_class=tc,
ready=ready,
checks=checks,
recommendation=recommendation,
)


def render_runtime_readiness_report(
report: RuntimeReadinessReport,
) -> str:
recommendation = report.recommendation
title = "READY" if report.ready else "NOT READY"
lines = [
f"Runtime readiness: {title}",
"",
f"Persona: {report.username} / {report.time_class}",
]
for check in report.checks:
suffix = ""
if check.path is not None:
suffix = f" ({check.path})"
if not check.required and not check.ok:
suffix += " [warning]"
lines.append(f"{check.name}: {check.status}{suffix}")
if check.detail:
lines.append(f" - {check.detail}")
lines.extend(
[
f"Recommended UI policy: {recommendation.move_policy_label}",
f"Move policy key: {recommendation.move_policy}",
]
)
return "\n".join(lines)


def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Check whether a ChessPersona runtime is ready for UI play."
)
parser.add_argument("--username", required=True)
parser.add_argument(
"--time-class",
choices=("rapid", "blitz"),
required=True,
)
parser.add_argument("--engine-path", default=str(DEFAULT_ENGINE_PATH))
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(
"--strict-metrics",
action="store_true",
help="Treat missing metrics/release-gate JSON as blocking.",
)
return parser


def run_from_args(args: argparse.Namespace) -> int:
report = build_runtime_readiness_report(
username=args.username,
time_class=args.time_class,
engine_path=args.engine_path,
persona_root=args.persona_root,
models_root=args.models_root,
evidence_root=args.evidence_root,
strict_metrics=bool(args.strict_metrics),
)
print(render_runtime_readiness_report(report))
return report.exit_code


def main(argv: Sequence[str] | None = None) -> int:
parser = build_arg_parser()
args = parser.parse_args(argv)
return run_from_args(args)


if __name__ == "__main__":
raise SystemExit(main())
Loading