diff --git a/src/ml/full_persona_pipeline.py b/src/ml/full_persona_pipeline.py index 758b2b9..61d0a1c 100644 --- a/src/ml/full_persona_pipeline.py +++ b/src/ml/full_persona_pipeline.py @@ -1432,3 +1432,784 @@ def read_release_gate_status(path: Path) -> FriendlyFinalStatus: ) return _previous_read_release_gate_status_step32a(path) + +# Step 32F: integrate elite evidence gate into full pipeline. +ELITE_GUARDED_SELFPLAY_DECISIONS = 320 +ELITE_GUARDED_SELFPLAY_GAMES = 8 +ELITE_GUARDED_SELFPLAY_PHASE_RECORDS = 20 +ELITE_EVIDENCE_ROOT = Path("data/ml/direct_ranker_elite_guarded_evidence") +ELITE_POLICY_GUARDED = "direct_ranker_elite_guarded" +ELITE_POLICY_GUARDED_STRICT = "direct_ranker_elite_guarded_strict" + + +def elite_policy_from_final_status(status: FriendlyFinalStatus) -> str: + combined = " ".join( + ( + status.status, + status.reason, + status.recommended_policy, + *tuple(str(note) for note in status.notes), + ) + ).lower() + + if "strict" in combined or "runtime.safety" in combined or "gap_over" in combined: + return ELITE_POLICY_GUARDED_STRICT + + return ELITE_POLICY_GUARDED + + +def elite_evidence_report_path( + username: str, + time_class: str, + root: Path = ELITE_EVIDENCE_ROOT, +) -> Path: + return ( + root + / normalize_username(username) + / f"{str(time_class).strip().lower()}_elite_guarded_evidence.json" + ) + + +def _direct_ranker_context_from_step( + step: PipelineStep, +) -> tuple[str, str, str]: + username = _option_value(step.command, "--username") + time_class = _option_value(step.command, "--time-class") + engine_path = _option_value(step.command, "--engine-path") + + if not username or not time_class: + raise ValueError( + "Tidak bisa menemukan --username/--time-class dari Direct Ranker command." + ) + + if not engine_path: + engine_path = str(DEFAULT_ENGINE_PATH) + + return ( + normalize_username(username), + str(time_class).strip().lower(), + str(engine_path), + ) + + +def build_elite_guarded_selfplay_step( + *, + status: FriendlyFinalStatus, + direct_ranker_step: PipelineStep, + python_bin: str | None = None, +) -> PipelineStep: + username, time_class, engine_path = _direct_ranker_context_from_step( + direct_ranker_step + ) + policy = elite_policy_from_final_status(status) + + return PipelineStep( + name="elite_guarded_selfplay", + title="Run elite guarded self-play evidence", + command=( + python_bin or sys.executable, + "run_direct_ranker_shadow_selfplay.py", + "--username", + username, + "--time-class", + time_class, + "--engine-path", + engine_path, + "--move-policy", + policy, + "--target-decisions", + str(ELITE_GUARDED_SELFPLAY_DECISIONS), + "--minimum-games", + str(ELITE_GUARDED_SELFPLAY_GAMES), + "--minimum-phase-records", + str(ELITE_GUARDED_SELFPLAY_PHASE_RECORDS), + "--strict-target", + "--reset-logs", + ), + notes=( + "Elite engine-like release gate block detected; running model-first guarded self-play evidence.", + ), + ) + + +def build_elite_guarded_evidence_step( + *, + status: FriendlyFinalStatus, + direct_ranker_step: PipelineStep, + python_bin: str | None = None, +) -> PipelineStep: + username, time_class, _engine_path = _direct_ranker_context_from_step( + direct_ranker_step + ) + policy = elite_policy_from_final_status(status) + + return PipelineStep( + name="elite_guarded_evidence_gate", + title="Evaluate elite guarded evidence gate", + command=( + python_bin or sys.executable, + "evaluate_direct_ranker_elite_guarded_evidence.py", + "--username", + username, + "--time-class", + time_class, + "--move-policy", + policy, + "--strict", + ), + notes=( + "Normal guarded beta remains separate; this gate approves elite model-first guarded runtime only.", + ), + ) + + +def read_elite_guarded_evidence_status( + *, + username: str, + time_class: str, + path: Path | None = None, +) -> FriendlyFinalStatus: + report_path = path or elite_evidence_report_path( + username=username, + time_class=time_class, + ) + payload = _load_json_object(report_path) + + if payload is None: + return FriendlyFinalStatus( + status="technical_failed", + decision="elite_evidence_report_unreadable", + reason="elite_evidence_report_unreadable", + recommended_policy="Statistical selector", + release_gate_path=report_path, + technical_error=True, + notes=(f"Elite evidence report tidak bisa dibaca: {report_path}",), + ) + + decision = str(payload.get("decision", "")).strip() + reason = str(payload.get("reason", "")).strip() or "unknown" + recommended_policy = ( + str(payload.get("recommended_policy", "")).strip() + or "Direct Ranker Elite Guarded after more evidence" + ) + + failed_checks_raw = payload.get("failed_checks", []) + failed_checks = ( + tuple(str(item) for item in failed_checks_raw) + if isinstance(failed_checks_raw, list) + else () + ) + + ready_decisions = { + "elite_guarded_ready", + "elite_guarded_strict_ready", + } + known_decisions = ready_decisions | { + "elite_guarded_blocked", + "needs_more_evidence", + } + + if decision in ready_decisions: + return FriendlyFinalStatus( + status=decision, + decision=decision, + reason=reason, + recommended_policy=recommended_policy, + release_gate_path=report_path, + technical_error=False, + notes=failed_checks, + ) + + if decision in known_decisions: + return FriendlyFinalStatus( + status=decision, + decision=decision, + reason=reason, + recommended_policy=recommended_policy, + release_gate_path=report_path, + technical_error=False, + notes=failed_checks, + ) + + return FriendlyFinalStatus( + status="technical_failed", + decision=decision or "unknown_elite_evidence_decision", + reason=reason, + recommended_policy="Statistical selector", + release_gate_path=report_path, + technical_error=True, + notes=failed_checks, + ) + + +def render_elite_guarded_pipeline_status( + status: FriendlyFinalStatus, +) -> str: + if status.status in { + "elite_guarded_ready", + "elite_guarded_strict_ready", + }: + title = "Pipeline completed with elite guarded evidence ready." + elif status.status == "needs_more_evidence": + title = "Pipeline completed, but elite guarded evidence needs more data." + elif status.status == "elite_guarded_blocked": + title = "Pipeline completed with elite guarded runtime blocked." + else: + title = "Pipeline finished with elite evidence technical failure." + + lines = [ + "", + "=" * 72, + title, + "", + "Final status:", + f"- status: {status.status}", + f"- decision: {status.decision}", + f"- reason: {status.reason}", + f"- recommended_policy: {status.recommended_policy}", + ] + + if status.release_gate_path is not None: + lines.append(f"- elite_evidence_report: {status.release_gate_path}") + + if status.notes: + lines.extend(["", "Details:"]) + for note in status.notes: + lines.append(f"- {note}") + + if status.status in { + "elite_guarded_ready", + "elite_guarded_strict_ready", + }: + lines.extend( + [ + "", + "Elite model-first guarded runtime is evidence-ready.", + "Normal Direct Ranker guarded beta release gate remains separate.", + ] + ) + else: + lines.extend( + [ + "", + "Ini safety/evidence block, bukan technical crash.", + "Model belum dianggap siap untuk elite guarded runtime sampai evidence gate lulus.", + ] + ) + + lines.append("=" * 72) + return "\n".join(lines).rstrip() + "\n" + + +def execute_elite_evidence_flow( + *, + status: FriendlyFinalStatus, + direct_ranker_step: PipelineStep, +) -> int: + username, time_class, _engine_path = _direct_ranker_context_from_step( + direct_ranker_step + ) + steps = ( + build_elite_guarded_selfplay_step( + status=status, + direct_ranker_step=direct_ranker_step, + ), + build_elite_guarded_evidence_step( + status=status, + direct_ranker_step=direct_ranker_step, + ), + ) + + for number, step in enumerate(steps, start=1): + print("\n" + "=" * 72) + print(f"[elite {number}/{len(steps)}] {step.name}: {step.title}") + print(format_command(step.command)) + for note in step.notes: + print(f"note: {note}") + print("=" * 72) + + completed = subprocess.run( + list(step.command), + check=False, + ) + + if completed.returncode != 0 and step.name != "elite_guarded_evidence_gate": + print( + f"\nElite evidence step gagal: {step.name} " + f"(exit code {completed.returncode})" + ) + return completed.returncode or 1 + + if step.name == "elite_guarded_evidence_gate": + evidence_status = read_elite_guarded_evidence_status( + username=username, + time_class=time_class, + ) + print( + render_elite_guarded_pipeline_status(evidence_status), + end="", + ) + + if evidence_status.technical_error: + return completed.returncode or 1 + + return 0 + + return 0 + + +# Step 32F override: after an elite engine-like release gate block, continue +# into elite self-play + elite evidence gate instead of stopping at the normal +# release gate status. +def execute_plan( + steps: Sequence[PipelineStep], + continue_on_error: bool = False, + refresh_step: Callable[[PipelineStep], PipelineStep] | None = None, +) -> int: + if not steps: + print("Tidak ada step yang dipilih.") + return 0 + + failures: list[tuple[str, int]] = [] + + for number, planned_step in enumerate( + steps, + start=1, + ): + step = ( + refresh_step(planned_step) + if refresh_step is not None + else planned_step + ) + + print("\n" + "=" * 72) + print(f"[{number}/{len(steps)}] {step.name}: {step.title}") + print(format_command(step.command)) + for note in step.notes: + print(f"note: {note}") + print("=" * 72) + + completed = subprocess.run( + list(step.command), + check=False, + ) + + if completed.returncode != 0: + friendly_status = is_nonfatal_release_gate_block( + step=step, + returncode=int(completed.returncode), + ) + + if friendly_status is not None: + if friendly_status.status == "elite_engine_like_persona": + print(render_final_status(friendly_status), end="") + return execute_elite_evidence_flow( + status=friendly_status, + direct_ranker_step=step, + ) + + print(render_final_status(friendly_status), end="") + return 0 + + failures.append( + ( + step.name, + completed.returncode, + ) + ) + print( + f"\nStep gagal: {step.name} " + f"(exit code {completed.returncode})" + ) + + if not continue_on_error: + return completed.returncode or 1 + + if failures: + print("\nPipeline selesai dengan error:") + for step_name, return_code in failures: + print(f"- {step_name}: exit code {return_code}") + return 1 + + print("\nPipeline selesai tanpa error.") + return 0 + +# Step 32F.1 hotfix: +# Direct Ranker persona pipeline may exit with code 1 even when the underlying +# failure is a valid release-gate block. Detect the release-gate report +# directly and continue elite engine-like cases into elite evidence flow. + +def _release_gate_failed_checks_from_report( + report: dict[str, Any], +) -> tuple[str, ...]: + candidates: list[str] = [] + + for key in ( + "failed_checks", + "blocking_failures", + "failures", + "failed", + ): + value = report.get(key) + if isinstance(value, list): + candidates.extend(str(item) for item in value) + elif isinstance(value, str): + candidates.append(value) + + strings = tuple(_iter_strings(report)) + for value in strings: + text = str(value).strip() + if "." in text and any( + prefix in text + for prefix in ( + "temporal.", + "runtime.", + "coverage.", + "model.", + "candidate.", + "shadow.", + ) + ): + candidates.append(text) + + deduped: list[str] = [] + for item in candidates: + clean = str(item).strip() + if clean and clean not in deduped: + deduped.append(clean) + + return tuple(deduped) + + +def _is_elite_engine_like_release_block( + *, + decision: str, + failed_checks: Sequence[str], + report: dict[str, Any], +) -> bool: + if decision != "release_candidate_blocked": + return False + + combined = " ".join( + ( + *tuple(str(item) for item in failed_checks), + *tuple(_iter_strings(report)), + ) + ).lower() + + return ( + "weak_temporal_persona_signal" in combined + or "temporal.quality.assessment" in combined + or "temporal.quality.vs_engine" in combined + ) + + +def _elite_recommended_policy_from_failed_checks( + failed_checks: Sequence[str], +) -> str: + combined = " ".join(str(item).lower() for item in failed_checks) + + if "runtime.safety" in combined or "gap_over" in combined: + return "Direct Ranker Elite Guarded Strict" + + return "Direct Ranker Elite Guarded" + + +def read_release_gate_status( + path: Path, +) -> FriendlyFinalStatus: + report = _load_json_object(path) + + if report is None: + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_report_unreadable", + reason="release_gate_report_unreadable", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=(f"Release gate report tidak bisa dibaca: {path}",), + ) + + decision = str(report.get("decision", "")).strip() + failed_checks = _release_gate_failed_checks_from_report(report) + + if decision == "release_candidate_ready": + return FriendlyFinalStatus( + status="release_candidate_ready", + decision=decision, + reason="release_gate_ready", + recommended_policy="Direct Ranker guarded beta", + release_gate_path=path, + technical_error=False, + notes=(), + ) + + if _is_elite_engine_like_release_block( + decision=decision, + failed_checks=failed_checks, + report=report, + ): + return FriendlyFinalStatus( + status="elite_engine_like_persona", + decision=decision, + reason="player_is_engine_like_high_accuracy", + recommended_policy=_elite_recommended_policy_from_failed_checks( + failed_checks + ), + release_gate_path=path, + technical_error=False, + notes=failed_checks, + ) + + if decision == "release_candidate_blocked": + return FriendlyFinalStatus( + status="trained_but_not_release_ready", + decision=decision, + reason=classify_release_gate_reason( + failed_checks, + tuple(_iter_strings(report)), + ), + recommended_policy="Direct Ranker shadow / Statistical selector", + release_gate_path=path, + technical_error=False, + notes=failed_checks, + ) + + return FriendlyFinalStatus( + status="technical_failed", + decision=decision or "unknown_release_gate_decision", + reason="unknown_release_gate_decision", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=failed_checks, + ) + + +def inspect_release_gate_status_for_command( + command: Sequence[str], +) -> FriendlyFinalStatus: + username = _option_value(command, "--username") + time_class = _option_value(command, "--time-class") + + if not username or not time_class: + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_context_missing", + reason="release_gate_context_missing", + recommended_policy="Statistical selector", + release_gate_path=None, + technical_error=True, + notes=("Tidak bisa menemukan --username/--time-class dari command.",), + ) + + path = release_gate_report_path( + username=username, + time_class=time_class, + ) + + if not path.exists(): + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_report_missing", + reason="release_gate_report_missing", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=(f"Release gate report tidak ditemukan: {path}",), + ) + + return read_release_gate_status(path) + + +def detect_final_status_from_release_gate(path: Path) -> FriendlyFinalStatus: + return read_release_gate_status(path) + +# Step 32F.3 hotfix: +# Some release-gate reports store the decision under nested assessment/status +# fields instead of a top-level "decision". Keep elite-flow detection robust +# across both report shapes. + +def _find_release_gate_decision_value(payload: Any) -> str: + if isinstance(payload, dict): + for key in ( + "decision", + "status", + "release_gate_status", + "release_status", + "final_decision", + ): + value = payload.get(key) + text = str(value or "").strip() + if text in { + "release_candidate_ready", + "release_candidate_blocked", + }: + return text + + for key in ( + "assessment", + "release_gate", + "gate", + "summary", + "final_status", + "result", + ): + value = payload.get(key) + nested = _find_release_gate_decision_value(value) + if nested: + return nested + + for value in payload.values(): + nested = _find_release_gate_decision_value(value) + if nested: + return nested + + elif isinstance(payload, list | tuple): + for item in payload: + nested = _find_release_gate_decision_value(item) + if nested: + return nested + + elif isinstance(payload, str): + text = payload.strip() + if text in { + "release_candidate_ready", + "release_candidate_blocked", + }: + return text + + return "" + + +def _release_gate_decision_from_report( + report: dict[str, Any], +) -> str: + decision = _find_release_gate_decision_value(report) + if decision: + return decision + + strings = tuple(_iter_strings(report)) + for value in strings: + text = str(value).strip() + if text in { + "release_candidate_ready", + "release_candidate_blocked", + }: + return text + + return "" + + +def read_release_gate_status( + path: Path, +) -> FriendlyFinalStatus: + report = _load_json_object(path) + + if report is None: + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_report_unreadable", + reason="release_gate_report_unreadable", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=(f"Release gate report tidak bisa dibaca: {path}",), + ) + + decision = _release_gate_decision_from_report(report) + failed_checks = _release_gate_failed_checks_from_report(report) + all_strings = tuple(_iter_strings(report)) + + if decision == "release_candidate_ready": + return FriendlyFinalStatus( + status="release_candidate_ready", + decision=decision, + reason="release_gate_ready", + recommended_policy="Direct Ranker guarded beta", + release_gate_path=path, + technical_error=False, + notes=(), + ) + + if _is_elite_engine_like_release_block( + decision=decision, + failed_checks=failed_checks, + report=report, + ): + return FriendlyFinalStatus( + status="elite_engine_like_persona", + decision=decision, + reason="player_is_engine_like_high_accuracy", + recommended_policy=_elite_recommended_policy_from_failed_checks( + failed_checks + ), + release_gate_path=path, + technical_error=False, + notes=failed_checks, + ) + + if decision == "release_candidate_blocked": + return FriendlyFinalStatus( + status="trained_but_not_release_ready", + decision=decision, + reason=classify_release_gate_reason( + failed_checks, + all_strings, + ), + recommended_policy="Direct Ranker shadow / Statistical selector", + release_gate_path=path, + technical_error=False, + notes=failed_checks, + ) + + return FriendlyFinalStatus( + status="technical_failed", + decision=decision or "unknown_release_gate_decision", + reason="unknown_release_gate_decision", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=failed_checks, + ) + + +def inspect_release_gate_status_for_command( + command: Sequence[str], +) -> FriendlyFinalStatus: + username = _option_value(command, "--username") + time_class = _option_value(command, "--time-class") + + if not username or not time_class: + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_context_missing", + reason="release_gate_context_missing", + recommended_policy="Statistical selector", + release_gate_path=None, + technical_error=True, + notes=("Tidak bisa menemukan --username/--time-class dari command.",), + ) + + path = release_gate_report_path( + username=username, + time_class=time_class, + ) + + if not path.exists(): + return FriendlyFinalStatus( + status="technical_failed", + decision="release_gate_report_missing", + reason="release_gate_report_missing", + recommended_policy="Statistical selector", + release_gate_path=path, + technical_error=True, + notes=(f"Release gate report tidak ditemukan: {path}",), + ) + + return read_release_gate_status(path) + + +def detect_final_status_from_release_gate(path: Path) -> FriendlyFinalStatus: + return read_release_gate_status(path) diff --git a/tests/test_full_persona_pipeline_elite_evidence.py b/tests/test_full_persona_pipeline_elite_evidence.py new file mode 100644 index 0000000..295eadd --- /dev/null +++ b/tests/test_full_persona_pipeline_elite_evidence.py @@ -0,0 +1,299 @@ +from __future__ import annotations + +import json +import types +from pathlib import Path + +from src.ml import full_persona_pipeline + + +def _direct_step() -> full_persona_pipeline.PipelineStep: + return full_persona_pipeline.PipelineStep( + name="direct_ranker_pipeline", + title="Run Direct Ranker pipeline", + command=( + "python", + "-m", + "src.ml.direct_ranker_persona_pipeline", + "--username", + "hikaru", + "--time-class", + "rapid", + "--engine-path", + "engines/stockfish/stockfish.exe", + "--execute", + "--yes", + ), + ) + + +def test_elite_policy_from_status_defaults_to_guarded() -> None: + status = full_persona_pipeline.FriendlyFinalStatus( + status="elite_engine_like_persona", + decision="release_candidate_blocked", + reason="player_is_engine_like_high_accuracy", + recommended_policy="Direct Ranker Elite Guarded", + release_gate_path=Path("release.json"), + ) + + assert ( + full_persona_pipeline.elite_policy_from_final_status(status) + == "direct_ranker_elite_guarded" + ) + + +def test_elite_policy_from_status_uses_strict_when_recommended() -> None: + status = full_persona_pipeline.FriendlyFinalStatus( + status="elite_engine_like_persona", + decision="release_candidate_blocked", + reason="player_is_engine_like_high_accuracy", + recommended_policy="Direct Ranker Elite Guarded Strict", + release_gate_path=Path("release.json"), + notes=("runtime.safety.gap_over_200_percent",), + ) + + assert ( + full_persona_pipeline.elite_policy_from_final_status(status) + == "direct_ranker_elite_guarded_strict" + ) + + +def test_build_elite_selfplay_step_uses_direct_context() -> None: + status = full_persona_pipeline.FriendlyFinalStatus( + status="elite_engine_like_persona", + decision="release_candidate_blocked", + reason="player_is_engine_like_high_accuracy", + recommended_policy="Direct Ranker Elite Guarded", + release_gate_path=Path("release.json"), + ) + + step = full_persona_pipeline.build_elite_guarded_selfplay_step( + status=status, + direct_ranker_step=_direct_step(), + python_bin="python", + ) + + assert step.name == "elite_guarded_selfplay" + assert "run_direct_ranker_shadow_selfplay.py" in step.command + assert "--reset-logs" in step.command + assert "direct_ranker_elite_guarded" in step.command + + +def test_execute_plan_runs_elite_evidence_flow_after_elite_block( + tmp_path: Path, + monkeypatch, +) -> None: + monkeypatch.chdir(tmp_path) + + direct_status = full_persona_pipeline.FriendlyFinalStatus( + status="elite_engine_like_persona", + decision="release_candidate_blocked", + reason="player_is_engine_like_high_accuracy", + recommended_policy="Direct Ranker Elite Guarded", + release_gate_path=Path("release.json"), + technical_error=False, + notes=(), + ) + + commands: list[tuple[str, ...]] = [] + + def fake_inspect(command): + return direct_status + + def fake_run(command: list[str], check: bool = False): + commands.append(tuple(command)) + + if "src.ml.direct_ranker_persona_pipeline" in command: + return types.SimpleNamespace(returncode=2) + + if "run_direct_ranker_shadow_selfplay.py" in command: + return types.SimpleNamespace(returncode=0) + + if "evaluate_direct_ranker_elite_guarded_evidence.py" in command: + report = ( + Path("data/ml/direct_ranker_elite_guarded_evidence") + / "hikaru" + / "rapid_elite_guarded_evidence.json" + ) + report.parent.mkdir(parents=True, exist_ok=True) + report.write_text( + json.dumps( + { + "decision": "elite_guarded_ready", + "reason": "elite_guarded_evidence_ready", + "recommended_policy": "Direct Ranker Elite Guarded", + "failed_checks": [], + } + ), + encoding="utf-8", + ) + return types.SimpleNamespace(returncode=0) + + return types.SimpleNamespace(returncode=1) + + monkeypatch.setattr( + full_persona_pipeline, + "inspect_release_gate_status_for_command", + fake_inspect, + ) + monkeypatch.setattr( + full_persona_pipeline.subprocess, + "run", + fake_run, + ) + + assert full_persona_pipeline.execute_plan([_direct_step()]) == 0 + + assert any("src.ml.direct_ranker_persona_pipeline" in cmd for cmd in commands) + assert any("run_direct_ranker_shadow_selfplay.py" in cmd for cmd in commands) + assert any( + "evaluate_direct_ranker_elite_guarded_evidence.py" in cmd + for cmd in commands + ) + + +def test_read_elite_guarded_evidence_status_ready( + tmp_path: Path, +) -> None: + path = tmp_path / "evidence.json" + path.write_text( + json.dumps( + { + "decision": "elite_guarded_strict_ready", + "reason": "elite_guarded_evidence_ready", + "recommended_policy": "Direct Ranker Elite Guarded Strict", + "failed_checks": [], + } + ), + encoding="utf-8", + ) + + status = full_persona_pipeline.read_elite_guarded_evidence_status( + username="hikaru", + time_class="blitz", + path=path, + ) + + assert status.status == "elite_guarded_strict_ready" + assert status.technical_ok is True + assert status.recommended_policy == "Direct Ranker Elite Guarded Strict" + +def test_read_release_gate_status_detects_full_elite_block_with_runtime_safety( + tmp_path: Path, +) -> None: + report_path = ( + tmp_path + / "data" + / "ml" + / "direct_candidate_release_gate" + / "hikaru" + / "rapid_release_gate.json" + ) + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text( + json.dumps( + { + "decision": "release_candidate_blocked", + "failed_checks": [ + "temporal.quality.assessment", + "temporal.quality.vs_engine", + "runtime.safety.gap_over_200_percent", + ], + "assessment": "weak_temporal_persona_signal", + } + ), + encoding="utf-8", + ) + + status = full_persona_pipeline.read_release_gate_status(report_path) + + assert status.status == "elite_engine_like_persona" + assert status.technical_ok is True + assert status.reason == "player_is_engine_like_high_accuracy" + assert ( + status.recommended_policy + == "Direct Ranker Elite Guarded Strict" + ) + + +def test_inspect_release_gate_status_for_command_detects_exit_one_elite_block( + tmp_path: Path, + monkeypatch, +) -> None: + monkeypatch.chdir(tmp_path) + + report_path = ( + Path("data/ml/direct_candidate_release_gate") + / "hikaru" + / "rapid_release_gate.json" + ) + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text( + json.dumps( + { + "decision": "release_candidate_blocked", + "failed_checks": [ + "temporal.quality.assessment", + "temporal.quality.vs_engine", + ], + } + ), + encoding="utf-8", + ) + + step = full_persona_pipeline.PipelineStep( + name="direct_ranker_pipeline", + title="Run Direct Ranker pipeline", + command=( + "python", + "-m", + "src.ml.direct_ranker_persona_pipeline", + "--username", + "hikaru", + "--time-class", + "rapid", + "--execute", + "--yes", + ), + ) + + status = full_persona_pipeline.is_nonfatal_release_gate_block( + step=step, + returncode=1, + ) + + assert status is not None + assert status.status == "elite_engine_like_persona" + +def test_read_release_gate_status_detects_nested_blocked_decision( + tmp_path: Path, +) -> None: + path = tmp_path / "nested_release_gate.json" + path.write_text( + json.dumps( + { + "assessment": { + "decision": "release_candidate_blocked", + }, + "checks": [ + {"id": "temporal.quality.assessment", "passed": False}, + {"id": "temporal.quality.vs_engine", "passed": False}, + { + "id": "runtime.safety.gap_over_200_percent", + "passed": False, + }, + ], + } + ), + encoding="utf-8", + ) + + status = full_persona_pipeline.read_release_gate_status(path) + + assert status.status == "elite_engine_like_persona" + assert status.decision == "release_candidate_blocked" + assert status.reason == "player_is_engine_like_high_accuracy" + assert ( + status.recommended_policy + == "Direct Ranker Elite Guarded Strict" + ) diff --git a/tests/test_full_persona_pipeline_final_status.py b/tests/test_full_persona_pipeline_final_status.py index c6f7679..a6aa5f3 100644 --- a/tests/test_full_persona_pipeline_final_status.py +++ b/tests/test_full_persona_pipeline_final_status.py @@ -30,7 +30,7 @@ def test_temporal_release_gate_block_is_friendly_status() -> None: ) -def test_execute_plan_treats_release_gate_block_as_nontechnical( +def test_execute_plan_elite_block_runs_evidence_flow_successfully( tmp_path: Path, monkeypatch, ) -> None: @@ -58,6 +58,30 @@ def fake_run(command: list[str], check: bool = False) -> object: ), encoding="utf-8", ) + + if "evaluate_direct_ranker_elite_guarded_evidence.py" in command: + evidence_path = ( + Path("data/ml/direct_ranker_elite_guarded_evidence") + / "hikaru" + / "rapid_elite_guarded_evidence.json" + ) + evidence_path.parent.mkdir(parents=True, exist_ok=True) + evidence_path.write_text( + json.dumps( + { + "decision": "elite_guarded_ready", + "reason": "elite_guarded_evidence_ready", + "recommended_policy": "Direct Ranker Elite Guarded", + "failed_checks": [], + } + ), + encoding="utf-8", + ) + return types.SimpleNamespace(returncode=0) + + if "run_direct_ranker_shadow_selfplay.py" in command: + return types.SimpleNamespace(returncode=0) + return types.SimpleNamespace(returncode=2) monkeypatch.setattr( diff --git a/tests/test_full_persona_pipeline_release_gate_block.py b/tests/test_full_persona_pipeline_release_gate_block.py index a96f661..569dc0e 100644 --- a/tests/test_full_persona_pipeline_release_gate_block.py +++ b/tests/test_full_persona_pipeline_release_gate_block.py @@ -2,24 +2,36 @@ import json import sys +import types from pathlib import Path from src.ml.full_persona_pipeline import ( PipelineStep, - classify_release_gate_reason, execute_plan, read_release_gate_status, - render_final_status, ) -def test_release_gate_block_is_friendly_status(tmp_path: Path) -> None: - report = tmp_path / "rapid_release_gate.json" +def _write_release_gate_report( + root: Path, + *, + failed_checks: list[str] | None = None, +) -> Path: + report = ( + root + / "data" + / "ml" + / "direct_candidate_release_gate" + / "hikaru" + / "rapid_release_gate.json" + ) + report.parent.mkdir(parents=True, exist_ok=True) report.write_text( json.dumps( { "decision": "release_candidate_blocked", - "failed_checks": [ + "failed_checks": failed_checks + or [ "temporal.quality.assessment", "temporal.quality.vs_engine", ], @@ -27,48 +39,47 @@ def test_release_gate_block_is_friendly_status(tmp_path: Path) -> None: ), encoding="utf-8", ) + return report - status = read_release_gate_status(report) - assert status.status == "trained_but_not_release_ready" - assert status.decision == "release_candidate_blocked" - assert status.reason == "weak_temporal_persona_signal" - assert status.recommended_policy == "Direct Ranker shadow / Statistical selector" - assert not status.technical_error - - -def test_render_blocked_release_gate_message_mentions_safety() -> None: - report = Path("data/ml/direct_candidate_release_gate/hikaru/rapid_release_gate.json") - status = read_release_gate_status.__globals__["FriendlyFinalStatus"]( - status="trained_but_not_release_ready", - decision="release_candidate_blocked", - reason="weak_temporal_persona_signal", - recommended_policy="Direct Ranker shadow / Statistical selector", - release_gate_path=report, - technical_error=False, - notes=("temporal.quality.assessment",), +def _write_elite_evidence_report(root: Path) -> Path: + report = ( + root + / "data" + / "ml" + / "direct_ranker_elite_guarded_evidence" + / "hikaru" + / "rapid_elite_guarded_evidence.json" ) + report.parent.mkdir(parents=True, exist_ok=True) + report.write_text( + json.dumps( + { + "decision": "elite_guarded_ready", + "reason": "elite_guarded_evidence_ready", + "recommended_policy": "Direct Ranker Elite Guarded", + "failed_checks": [], + } + ), + encoding="utf-8", + ) + return report - rendered = render_final_status(status) - - assert "Pipeline completed with blocked release gate." in rendered - assert "trained_but_not_release_ready" in rendered - assert "Ini safety block, bukan technical crash." in rendered +def test_release_gate_block_is_friendly_status(tmp_path: Path) -> None: + report = _write_release_gate_report(tmp_path) -def test_classify_runtime_safety_reason() -> None: - assert ( - classify_release_gate_reason(["runtime.safety.gap_over_200_percent"]) - == "runtime_safety_not_ready" - ) + status = read_release_gate_status(report) + assert status.status == "elite_engine_like_persona" + assert status.decision == "release_candidate_blocked" + assert status.reason == "player_is_engine_like_high_accuracy" + assert status.technical_ok is True + assert status.exit_code == 0 + assert status.recommended_policy == "Direct Ranker Elite Guarded" -def test_execute_plan_treats_direct_ranker_block_as_nonfatal( - tmp_path: Path, - monkeypatch, -) -> None: - monkeypatch.chdir(tmp_path) +def test_release_gate_ready_is_release_candidate(tmp_path: Path) -> None: report = ( tmp_path / "data" @@ -77,20 +88,50 @@ def test_execute_plan_treats_direct_ranker_block_as_nonfatal( / "hikaru" / "rapid_release_gate.json" ) - report.parent.mkdir(parents=True) + report.parent.mkdir(parents=True, exist_ok=True) report.write_text( json.dumps( { - "decision": "release_candidate_blocked", - "failed_checks": [ - "temporal.quality.assessment", - "temporal.quality.vs_engine", - ], + "decision": "release_candidate_ready", + "failed_checks": [], } ), encoding="utf-8", ) + status = read_release_gate_status(report) + + assert status.status == "release_candidate_ready" + assert status.technical_ok is True + assert status.recommended_policy == "Direct Ranker guarded beta" + + +def test_execute_plan_treats_direct_ranker_block_as_nonfatal( + tmp_path: Path, + monkeypatch, +) -> None: + monkeypatch.chdir(tmp_path) + _write_release_gate_report(tmp_path) + + commands: list[tuple[str, ...]] = [] + + def fake_run(command: list[str], check: bool = False) -> object: + commands.append(tuple(command)) + + if "run_direct_ranker_shadow_selfplay.py" in command: + return types.SimpleNamespace(returncode=0) + + if "evaluate_direct_ranker_elite_guarded_evidence.py" in command: + _write_elite_evidence_report(tmp_path) + return types.SimpleNamespace(returncode=0) + + return types.SimpleNamespace(returncode=1) + + monkeypatch.setattr( + "src.ml.full_persona_pipeline.subprocess.run", + fake_run, + ) + step = PipelineStep( name="direct_ranker_pipeline", title="Run Direct Ranker pipeline", @@ -106,3 +147,19 @@ def test_execute_plan_treats_direct_ranker_block_as_nonfatal( ) assert execute_plan([step]) == 0 + assert any("run_direct_ranker_shadow_selfplay.py" in cmd for cmd in commands) + assert any( + "evaluate_direct_ranker_elite_guarded_evidence.py" in cmd + for cmd in commands + ) + + +def test_release_gate_unreadable_is_technical_failure(tmp_path: Path) -> None: + report = tmp_path / "broken_release_gate.json" + report.write_text("{not json", encoding="utf-8") + + status = read_release_gate_status(report) + + assert status.status == "technical_failed" + assert status.technical_ok is False + assert status.exit_code == 1