codec.py

1"""Versioned standard-chess state prompt; shared by preparation and inference."""
2import hashlib,importlib.util,sys
3from pathlib import Path
4if importlib.util.find_spec('chess') is None:
5    p=Path('/work/.cache/pypoetry/virtualenvs/rulebound-LLUAbKSL-py3.11/lib/python3.11/site-packages/chess')
6    s=importlib.util.spec_from_file_location('chess',p/'__init__.py',submodule_search_locations=[str(p)])
7    m=importlib.util.module_from_spec(s);sys.modules['chess']=m;s.loader.exec_module(m)
8import chess
9
10VERSION='chess-state-history16-v2'
11BOS,EOS,PAD=69,70,71 # Legacy meanings/IDs retained; EOS is never emitted here.
12PIECE_BASE=128 # empty, white P/N/B/R/Q/K, black P/N/B/R/Q/K
13SIDE_BASE=141;CASTLE_BASE=143;EP_NONE=159;EP_BASE=160
14HALF_BASE=224;HALF_GE150=374;HIST_NONE=375;HIST_PARTIAL=376;STATE=377;NEXT=378
15VOCAB=512;HISTORY=16;PROMPT=120;INPUT=122;TARGET_POSITIONS=(119,120,121)
16PROMOTION={None:64,chess.KNIGHT:65,chess.BISHOP:66,chess.ROOK:67,chess.QUEEN:68}
17REVERSE={v:k for k,v in PROMOTION.items()}
18CONTRACT={'version':VERSION,'vocab':VOCAB,'prompt_positions':PROMPT,'model_positions':INPUT,
19    'serialized_positions':123,'target_positions':TARGET_POSITIONS,'history_plies':HISTORY,
20    'board_order':'a1..h8','ep':'raw FEN target, including non-capturable target',
21    'halfmove':'exact0..149; GE150 terminal range','fullmove':'metadata only',
22    'history':'genuine suffix of authoritative move stack; left PAD keys masked; fixed position IDs',
23    'legacy_reserved':'72..127 remain reserved','new_used':'128..378','remaining_reserved':'379..511',
24    'loss':'three move targets only; no EOS or prefix reconstruction'}
25
26def check_board(board):
27    if type(board) is not chess.Board or board.chess960 or not board.is_valid():raise ValueError('valid standard chess.Board required')
28    if type(board.halfmove_clock) is not int or board.halfmove_clock<0 or type(board.fullmove_number) is not int or board.fullmove_number<1:raise ValueError('invalid clock metadata')
29
30def triple(move):
31    if not move or move.from_square == move.to_square or move.drop or move.promotion not in PROMOTION:raise ValueError('null/drop/unsupported move')
32    return [move.from_square,move.to_square,PROMOTION[move.promotion]]
33
34def decode(values):
35    if len(values)!=3 or any(type(x) is not int for x in values) or not all(0<=x<64 for x in values[:2]) or values[2] not in REVERSE:raise ValueError('invalid move triple')
36    move=chess.Move(values[0],values[1],promotion=REVERSE[values[2]])
37    if not move or move.from_square == move.to_square:raise ValueError('null/equal-square move')
38    return move
39
40def castle(board):
41    return sum(bit for bit,ok in ((1,board.has_kingside_castling_rights(chess.WHITE)),(2,board.has_queenside_castling_rights(chess.WHITE)),(4,board.has_kingside_castling_rights(chess.BLACK)),(8,board.has_queenside_castling_rights(chess.BLACK))) if ok)
42
43def signature(board):return board.fen(en_passant='fen')
44
45def position_key(board):
46    """Conservative transposition key: ignore clocks/history; legal EP semantics.
47
48    Raw unreachable EP targets collapse with no-EP positions. Distinct castling
49    rights and side-to-move remain distinct. SHA256 collisions cannot admit a leak.
50    """
51    check_board(board)
52    fields=board.fen(en_passant='legal').split()[:4]
53    return hashlib.sha256(' '.join(fields).encode('ascii')).digest()
54
55def genuine_history(board,n=HISTORY):
56    if type(n) is not int or not 0<=n<=HISTORY:raise ValueError('history bound')
57    n=min(n,len(board.move_stack));prior=board.copy(stack=True);moves=[]
58    for _ in range(n):moves.append(prior.pop())
59    moves.reverse()
60    for move in moves:
61        if move not in prior.legal_moves:raise ValueError('invalid history stack')
62        prior.push(move)
63    if signature(prior)!=signature(board):raise ValueError('history does not reach supplied board')
64    return moves
65
66def prompt(board,history_plies=HISTORY):
67    check_board(board);moves=genuine_history(board,history_plies)
68    missing=3*(HISTORY-len(moves))
69    ids=[BOS,HIST_PARTIAL if moves else HIST_NONE]+[PAD]*missing
70    for move in moves:ids.extend(triple(move))
71    ids.append(STATE)
72    for square in chess.SQUARES:
73        piece=board.piece_at(square)
74        ids.append(PIECE_BASE+(0 if piece is None else piece.piece_type+(0 if piece.color else 6)))
75    ids.extend([SIDE_BASE+(0 if board.turn else 1),CASTLE_BASE+castle(board),EP_NONE if board.ep_square is None else EP_BASE+board.ep_square,
76        HALF_BASE+board.halfmove_clock if board.halfmove_clock<150 else HALF_GE150,NEXT])
77    mask=[1,1]+[0]*missing+[1]*(PROMPT-2-missing)
78    assert len(ids)==len(mask)==PROMPT
79    return {'input_ids':ids,'attention_mask':mask,'position_ids':list(range(PROMPT)),
80        'metadata':{'fullmove_number':board.fullmove_number,'history_plies':len(moves),'halfmove_clock':board.halfmove_clock}}
81
82def from_history(start_fen,uci_moves):
83    board=chess.Board(start_fen);check_board(board)
84    if len(uci_moves)>HISTORY:raise ValueError('history exceeds16 plies')
85    for uci in uci_moves:
86        move=chess.Move.from_uci(uci)
87        if not move or move not in board.legal_moves:raise ValueError('illegal supplied history')
88        board.push(move)
89    return board
90
91def example(board,move,history_plies=HISTORY):
92    check_board(board)
93    if board.is_game_over(claim_draw=False):raise ValueError('automatic terminal position has no training decision')
94    if move not in board.legal_moves:raise ValueError('illegal target')
95    p=prompt(board,history_plies);target=triple(move)
96    return {'input_ids':p['input_ids']+target[:2],'attention_mask':p['attention_mask']+[1,1],
97        'position_ids':list(range(INPUT)),'labels':target,'loss_mask':[int(i in TARGET_POSITIONS) for i in range(INPUT)],
98        'metadata':p['metadata'],'position_key':position_key(board)}
99
100def masks(board,prefix=(),legal=False):
101    """Separate optional output constraints. Raw512 logits are never renormalized here."""
102    check_board(board)
103    if len(prefix)>2 or any(type(x) is not int or not 0<=x<64 for x in prefix):raise ValueError('incorrect move phase')
104    allowed=set(range(64) if len(prefix)<2 else range(64,69))
105    if legal:
106        allowed={triple(m)[len(prefix)] for m in board.legal_moves if triple(m)[:len(prefix)]==list(prefix)}
107    return [i in allowed for i in range(VOCAB)]
108