""" EXPERIMENTAL — OAM vs Conventional Engine Criticality ====================================================== Purpose (falsifiable): Determine whether OAM's Moment detector is substantially reducible to conventional engine criticality on a small corpus. Proposition under test: "OAM's current Moment detector is substantially reducible to conventional engine criticality." Report exactly one of: SUPPORTED · WEAKENED · INSUFFICIENT EVIDENCE Rules honoured: - Production OAM code (`/app/backend/oam/`, `oam_analyze.py`, `frontend/src/pgn/parser.js`) is imported verbatim; not modified. - Stockfish is used only as an independent baseline; it is not combined with OAM. - No psychology, intention, or hidden state is introduced. - Reported evidence traces to (game, ply, OAM output, engine output). This file is EXPERIMENTAL. Not part of production surfaces. """ from __future__ import annotations import json import os import statistics import sys from dataclasses import dataclass, asdict from typing import List, Optional, Dict # Bring the production OAM package into scope without modification. sys.path.insert(0, "/app/backend") import chess # noqa: E402 import chess.engine # noqa: E402 import chess.pgn # noqa: E402 import io # noqa: E402 from oam import produce_candidates # noqa: E402 # ---------------------------------------------------------------------- # Corpus. Five short classical games + the frozen Evergreen fixture. # Every PGN below is a published historical game; no invented players. # ---------------------------------------------------------------------- def _load_repo_sample_pgn() -> str: """Read the Evergreen fixture from the frontend parser verbatim.""" path = "/app/frontend/src/pgn/parser.js" with open(path, "r", encoding="utf-8") as fh: src = fh.read() start = src.index("const SAMPLE_PGN = `") + len("const SAMPLE_PGN = `") end = src.index("`;", start) return src[start:end] CORPUS: List[dict] = [ { "id": "evergreen_1852", "name": "Anderssen–Dufresne, 1852 (Evergreen)", "pgn": _load_repo_sample_pgn(), "note": "Mandatory §9 fixture. OAM invariant Ply 34 · Black · gxf6 · OE 6/8.", }, # Morphy — Duke of Brunswick & Count Isouard, 1858 (Opera Game). { "id": "opera_1858", "name": "Morphy vs Duke Karl & Count Isouard, 1858 (Opera Game)", "pgn": ( '[Event "Paris"]\n[Site "Paris"]\n[Date "1858.??.??"]\n' '[White "Morphy, Paul"]\n[Black "Duke Karl / Count Isouard"]\n' '[Result "1-0"]\n\n' "1. e4 e5 2. Nf3 d6 3. d4 Bg4 4. dxe5 Bxf3 5. Qxf3 dxe5 6. Bc4 Nf6 " "7. Qb3 Qe7 8. Nc3 c6 9. Bg5 b5 10. Nxb5 cxb5 11. Bxb5+ Nbd7 12. O-O-O Rd8 " "13. Rxd7 Rxd7 14. Rd1 Qe6 15. Bxd7+ Nxd7 16. Qb8+ Nxb8 17. Rd8# 1-0" ), }, # Kasparov vs Topalov, Wijk aan Zee 1999. { "id": "kasparov_topalov_1999", "name": "Kasparov vs Topalov, Wijk aan Zee 1999", "pgn": ( '[Event "Hoogovens"]\n[Site "Wijk aan Zee NED"]\n[Date "1999.01.20"]\n' '[White "Kasparov, Garry"]\n[Black "Topalov, Veselin"]\n' '[Result "1-0"]\n\n' "1. e4 d6 2. d4 Nf6 3. Nc3 g6 4. Be3 Bg7 5. Qd2 c6 6. f3 b5 " "7. Nge2 Nbd7 8. Bh6 Bxh6 9. Qxh6 Bb7 10. a3 e5 11. O-O-O Qe7 " "12. Kb1 a6 13. Nc1 O-O-O 14. Nb3 exd4 15. Rxd4 c5 16. Rd1 Nb6 " "17. g3 Kb8 18. Na5 Ba8 19. Bh3 d5 20. Qf4+ Ka7 21. Rhe1 d4 " "22. Nd5 Nbxd5 23. exd5 Qd6 24. Rxd4 cxd4 25. Re7+ Kb6 " "26. Qxd4+ Kxa5 27. b4+ Ka4 28. Qc3 Qxd5 29. Ra7 Bb7 30. Rxb7 Qc4 " "31. Qxf6 Kxa3 32. Qxa6+ Kxb4 33. c3+ Kxc3 34. Qa1+ Kd2 " "35. Qb2+ Kd1 36. Bf1 Rd2 37. Rd7 Rxd7 38. Bxc4 bxc4 " "39. Qxh8 Rd3 40. Qa8 c3 41. Qa4+ Ke1 42. f4 f5 43. Kc1 Rd2 " "44. Qa7 1-0" ), }, # A short technical draw (Capablanca) — expected quiet. { "id": "capablanca_marshall_1918", "name": "Capablanca vs Marshall, New York 1918 (Marshall Attack origin — shortened)", "pgn": ( '[Event "New York"]\n[Site "New York"]\n[Date "1918.??.??"]\n' '[White "Capablanca, Jose Raul"]\n[Black "Marshall, Frank"]\n' '[Result "1-0"]\n\n' "1. e4 e5 2. Nf3 Nc6 3. Bb5 a6 4. Ba4 Nf6 5. O-O d5 6. exd5 e4 " "7. dxc6 exf3 8. Qxf3 Qe7 9. Bxf7+ Kxf7 10. Qxf3+ Kg8 " "11. Nc3 O-O-O 12. d3 Bb4 13. Bd2 Qe6 14. Rae1 Qxa2 " "15. Nd5 Nxd5 16. Qxd5 Rd6 17. Qc5 Rd5 18. Qe3 Qxe3 " "19. Bxe3 Rd6 20. Rxe6 Rxe6 1-0" ), }, # Byrne–Fischer 1956 ("Game of the Century") — expected sharp. { "id": "byrne_fischer_1956", "name": "D. Byrne vs Fischer, New York 1956 (Game of the Century)", "pgn": ( '[Event "Rosenwald"]\n[Site "New York"]\n[Date "1956.10.17"]\n' '[White "Byrne, Donald"]\n[Black "Fischer, Robert James"]\n' '[Result "0-1"]\n\n' "1. Nf3 Nf6 2. c4 g6 3. Nc3 Bg7 4. d4 O-O 5. Bf4 d5 6. Qb3 dxc4 " "7. Qxc4 c6 8. e4 Nbd7 9. Rd1 Nb6 10. Qc5 Bg4 11. Bg5 Na4 " "12. Qa3 Nxc3 13. bxc3 Nxe4 14. Bxe7 Qb6 15. Bc4 Nxc3 16. Bc5 Rfe8+ " "17. Kf1 Be6 18. Bxb6 Bxc4+ 19. Kg1 Ne2+ 20. Kf1 Nxd4+ " "21. Kg1 Ne2+ 22. Kf1 Nc3+ 23. Kg1 axb6 24. Qb4 Ra4 25. Qxb6 Nxd1 " "26. h3 Rxa2 27. Kh2 Nxf2 28. Re1 Rxe1 29. Qd8+ Bf8 30. Nxe1 Bd5 " "31. Nf3 Ne4 32. Qb8 b5 33. h4 h5 34. Ne5 Kg7 35. Kg1 Bc5+ " "36. Kf1 Ng3+ 37. Ke1 Bb4+ 38. Kd1 Bb3+ 39. Kc1 Ne2+ " "40. Kb1 Nc3+ 41. Kc1 Rc2# 0-1" ), }, # A short caro-kann miniature — quieter positional feel. { "id": "botvinnik_capa_1938", "name": "Botvinnik vs Capablanca, AVRO 1938 (short excerpt via first 30 plies)", "pgn": ( '[Event "AVRO"]\n[Site "Netherlands"]\n[Date "1938.11.??"]\n' '[White "Botvinnik, Mikhail"]\n[Black "Capablanca, Jose Raul"]\n' '[Result "1-0"]\n\n' "1. d4 Nf6 2. c4 e6 3. Nc3 Bb4 4. e3 d5 5. a3 Bxc3+ 6. bxc3 c5 " "7. cxd5 exd5 8. Bd3 O-O 9. Ne2 b6 10. O-O Ba6 11. Bxa6 Nxa6 " "12. Bb2 Qd7 13. a4 Rfe8 14. Qd3 c4 15. Qc2 Nb8 16. Rae1 Nc6 " "17. Ng3 Na5 18. f3 Nb3 19. e4 Qxa4 20. e5 Nd7 21. Qf2 g6 " "22. f4 f5 23. exf6 Nxf6 24. f5 Rxe1 25. Rxe1 Re8 26. Re6 Rxe6 " "27. fxe6 Kg7 28. Qf4 Qe8 29. Qe5 Qe7 30. Ba3 Qxa3 31. Nh5+ gxh5 " "32. Qg5+ Kf8 33. Qxf6+ Kg8 34. e7 Qc1+ 35. Kf2 Qc2+ " "36. Kg3 Qd3+ 37. Kh4 Qe4+ 38. Kxh5 Qe2+ 39. Kh4 Qe4+ " "40. g4 Qe1+ 41. Kh5 1-0" ), }, ] # ---------------------------------------------------------------------- # Stockfish per-ply evaluation # ---------------------------------------------------------------------- STOCKFISH_PATH = "/usr/games/stockfish" STOCKFISH_DEPTH = 16 # bounded fixed-depth for reproducibility @dataclass class PlyEval: ply: int # 1-indexed, matches OAM's response_ply convention color: str # 'w' or 'b' — the side that just moved played_san: str played_uci: str eval_before_cp: Optional[int] # cp from side-to-move perspective before the move (POV: mover) eval_after_cp: Optional[int] # cp from side-to-move perspective after their move (POV: opponent) eval_swing_cp: Optional[int] # abs change, normalised to side-to-move POV best_move_before: Optional[str] # engine best move at pre-move position (SAN) played_matched_best: Optional[bool] def _cp_from_pov(score: chess.engine.PovScore, pov_color: chess.Color) -> Optional[int]: """Return centipawn value from the requested colour's point of view. Mate is clamped to ±10000 to keep the swing metric bounded.""" s = score.pov(pov_color) if s.is_mate(): mate = s.mate() if mate is None: return None return 10000 if mate > 0 else -10000 return s.score() def per_ply_engine_metrics(pgn_text: str, depth: int = STOCKFISH_DEPTH) -> List[PlyEval]: game = chess.pgn.read_game(io.StringIO(pgn_text)) if game is None: raise ValueError("PGN parse failed") board = game.board() plies: List[PlyEval] = [] engine = chess.engine.SimpleEngine.popen_uci(STOCKFISH_PATH) try: ply_number = 0 for move in game.mainline_moves(): ply_number += 1 mover_color = board.turn # side to move BEFORE this move san_before = board.san(move) uci = move.uci() # Eval before the move — from mover's POV. info_before = engine.analyse(board, chess.engine.Limit(depth=depth)) eval_before = _cp_from_pov(info_before["score"], mover_color) best_move_before_uci = None best_move_before_san = None pv = info_before.get("pv") if pv: best_move_before_uci = pv[0].uci() best_move_before_san = board.san(pv[0]) played_matched_best = ( best_move_before_uci == uci if best_move_before_uci is not None else None ) # Play the move. board.push(move) # Eval after — still from ORIGINAL mover's POV so swing is comparable. info_after = engine.analyse(board, chess.engine.Limit(depth=depth)) eval_after = _cp_from_pov(info_after["score"], mover_color) if eval_before is not None and eval_after is not None: swing = abs(eval_after - eval_before) else: swing = None plies.append( PlyEval( ply=ply_number, color="w" if mover_color == chess.WHITE else "b", played_san=san_before, played_uci=uci, eval_before_cp=eval_before, eval_after_cp=eval_after, eval_swing_cp=swing, best_move_before=best_move_before_san, played_matched_best=played_matched_best, ) ) finally: engine.quit() return plies # ---------------------------------------------------------------------- # OAM per-game analysis (production pipeline, unchanged) # ---------------------------------------------------------------------- @dataclass class OamMoment: ply: int color: str response_san: str fts: int rds: int rls: int oe_score: int trigger_ply: int def oam_qualifying_moments(pgn_text: str, game_id: str) -> List[OamMoment]: candidates = produce_candidates(pgn_text, game_id=game_id, record_scope="COMPLETE_GAME") # Derive SAN per response_ply from the PGN (production frontend uses # moves[response_ply - 1]). game = chess.pgn.read_game(io.StringIO(pgn_text)) board = game.board() san_by_ply: Dict[int, str] = {} for i, mv in enumerate(game.mainline_moves(), start=1): san_by_ply[i] = board.san(mv) board.push(mv) out: List[OamMoment] = [] for c in candidates: if c.outcome != "QUALIFYING_CANDIDATE": continue out.append( OamMoment( ply=c.response_ply, color=c.player_side, response_san=san_by_ply.get(c.response_ply, "?"), fts=c.fts, rds=c.rds, rls=c.rls, oe_score=c.oe_score, trigger_ply=c.trigger_ply, ) ) return out # ---------------------------------------------------------------------- # Engine-critical rule # ---------------------------------------------------------------------- def engine_critical_plies(evals: List[PlyEval]) -> List[int]: """Deterministic rule (§3): ply(ies) with maximum absolute eval swing. Ties preserved rather than broken arbitrarily.""" scored = [(p.ply, p.eval_swing_cp) for p in evals if p.eval_swing_cp is not None] if not scored: return [] max_swing = max(s for _, s in scored) return [ply for ply, s in scored if s == max_swing] def evaluation_stable_at_ply(evals: List[PlyEval], ply: int, threshold_cp: int = 50) -> bool: for e in evals: if e.ply == ply: return e.eval_swing_cp is not None and e.eval_swing_cp < threshold_cp return False # ---------------------------------------------------------------------- # Classification (A/B/C/D) # ---------------------------------------------------------------------- def classify(oam: List[OamMoment], engine_crit: List[int]) -> Dict[str, List]: """Per §6: classify each OAM Moment and note any engine-only events. We return a per-Moment classification list plus an engine-only list of critical plies not selected by any OAM Moment.""" per_moment = [] oam_plies = {m.ply for m in oam} for m in oam: if m.ply in engine_crit: per_moment.append(("A_CONVERGENCE", m.ply, engine_crit)) elif engine_crit: per_moment.append(("B_DIVERGENCE", m.ply, engine_crit)) else: per_moment.append(("C_OAM_ONLY", m.ply, engine_crit)) engine_only = [p for p in engine_crit if p not in oam_plies] return {"per_moment": per_moment, "engine_only_plies": engine_only} # ---------------------------------------------------------------------- # Main # ---------------------------------------------------------------------- def main() -> None: out_dir = "/app/experiments/results" os.makedirs(out_dir, exist_ok=True) all_games = [] all_a = 0 all_b = 0 all_c = 0 all_d_events = 0 correlation_pairs: List[tuple] = [] # (OE_SCORE, engine swing at same ply) for game in CORPUS: print(f"\n== {game['name']} ==", flush=True) print("Running OAM...", flush=True) oam = oam_qualifying_moments(game["pgn"], game_id=game["id"]) print(f" OAM qualifying Moments: {len(oam)}", flush=True) print(f"Running Stockfish depth={STOCKFISH_DEPTH}...", flush=True) try: evals = per_ply_engine_metrics(game["pgn"], depth=STOCKFISH_DEPTH) except Exception as e: print(f" ENGINE ERROR: {e}", flush=True) evals = [] print(f" plies analysed: {len(evals)}", flush=True) engine_crit = engine_critical_plies(evals) cls = classify(oam, engine_crit) # Distance between each OAM Moment and the nearest engine-critical ply for m in oam: if engine_crit: nearest = min(engine_crit, key=lambda p: abs(p - m.ply)) distance = abs(nearest - m.ply) else: nearest = None distance = None # correlation datum: OE vs eval swing at the OAM ply oam_ply_swing = next((e.eval_swing_cp for e in evals if e.ply == m.ply), None) correlation_pairs.append((m.oe_score, oam_ply_swing)) # Convergence / divergence counts for tag, _, _ in cls["per_moment"]: if tag == "A_CONVERGENCE": all_a += 1 elif tag == "B_DIVERGENCE": all_b += 1 elif tag == "C_OAM_ONLY": all_c += 1 all_d_events += len(cls["engine_only_plies"]) all_games.append({ "id": game["id"], "name": game["name"], "note": game.get("note"), "oam_moments": [asdict(m) for m in oam], "engine_critical_plies": engine_crit, "engine_max_swing_cp": ( max((e.eval_swing_cp for e in evals if e.eval_swing_cp is not None), default=None) ), "n_plies": len(evals), "per_moment_classification": [ { "tag": tag, "oam_ply": ply, "engine_critical_plies": crit, "eval_swing_cp_at_oam_ply": next( (e.eval_swing_cp for e in evals if e.ply == ply), None ), } for tag, ply, crit in cls["per_moment"] ], "engine_only_plies": [ {"ply": p, "swing_cp": next((e.eval_swing_cp for e in evals if e.ply == p), None), "played_san": next((e.played_san for e in evals if e.ply == p), None), "engine_best_before": next((e.best_move_before for e in evals if e.ply == p), None), "played_matched_best": next((e.played_matched_best for e in evals if e.ply == p), None)} for p in cls["engine_only_plies"] ], "per_ply_engine": [asdict(e) for e in evals], }) # Correlation between OE_SCORE and engine swing at OAM plies pairs = [(oe, sw) for (oe, sw) in correlation_pairs if oe is not None and sw is not None] corr = None if len(pairs) >= 3: oes = [float(a) for a, _ in pairs] sws = [float(b) for _, b in pairs] try: corr = statistics.correlation(oes, sws) except statistics.StatisticsError: corr = None summary = { "config": { "engine": { "binary": STOCKFISH_PATH, "version": "Stockfish 15.1", "depth": STOCKFISH_DEPTH, "eval_convention": "centipawns, POV = side that moved (bounded ±10000 on mate)", "criticality_rule": "argmax of |eval_after - eval_before| per ply; ties preserved", }, "oam": { "package": "/app/backend/oam", "call": "produce_candidates(pgn, game_id, record_scope='COMPLETE_GAME')", "selection_rule": "outcome == QUALIFYING_CANDIDATE", "modified": False, }, "corpus_size": len(CORPUS), "games": [{"id": g["id"], "name": g["name"]} for g in CORPUS], }, "classification_totals": { "A_CONVERGENCE_moments": all_a, "B_DIVERGENCE_moments": all_b, "C_OAM_ONLY_moments": all_c, "D_ENGINE_ONLY_events": all_d_events, }, "correlation_oe_vs_engine_swing_at_OAM_plies": { "n_pairs": len(pairs), "pearson_r": corr, }, "games": all_games, } out_path = os.path.join(out_dir, "oam_vs_engine.json") with open(out_path, "w", encoding="utf-8") as fh: json.dump(summary, fh, indent=2) print(f"\nWrote {out_path}") if __name__ == "__main__": main()