core.py

1"""Public bridge logic. No credentials, local paths or HTTP on import."""
2import hashlib,json,time,re
3from pathlib import Path
4import chess,chess.pgn
5
6def identity(user):return str(user.get('id') or user.get('name') or '').lower()
7
8def challenge_reason(c,opponent,bot,busy=False):
9    if busy:return 'later'
10    if identity(c.get('challenger',{}))!=opponent.lower():return 'generic'
11    if identity(c.get('destUser',{}))!=bot.lower():return 'generic'
12    if c.get('rated') is not False:return 'casual'
13    if c.get('variant',{}).get('key')!='standard':return 'standard'
14    if c.get('initialFen') not in (None,'startpos',chess.STARTING_FEN):return 'standard'
15    tc=c.get('timeControl',{})
16    if tc.get('type')!='clock' or tc.get('limit',0)<600 or tc.get('increment',0)<5:return 'tooFast'
17    return None
18
19def reconstruct(moves):
20    board=chess.Board()
21    for text in moves.split():
22        move=board.parse_uci(text) # normalizes king-to-rook castling notation
23        if move not in board.legal_moves:raise ValueError('illegal server history')
24        board.push(move)
25    return board
26
27def position_key(board):
28    return hashlib.sha256(' '.join(m.uci() for m in board.move_stack).encode()).hexdigest()
29
30def budget(state,color,age=0):
31    side='w' if color else 'b';remaining=max(0,state[side+'time']/1000-age)
32    increment=state.get(side+'inc',0)/1000
33    seconds=max(.01,min(8.,remaining/30+increment*.5,remaining-3.))
34    expiration=state.get('expiration',{})
35    if expiration:
36        seconds=max(.01,min(seconds,(expiration.get('millisToMove',0)-expiration.get('idleMillis',0))/1000-age-2.))
37    return seconds
38
39def atomic_json(path,value):
40    tmp=path.with_suffix('.tmp');tmp.write_text(json.dumps(value,indent=2));tmp.replace(path)
41
42class Game:
43    def __init__(self,full,bot,opponent,directory):
44        if full.get('variant',{}).get('key')!='standard' or full.get('rated') is not False:
45            raise ValueError('game outside allowed variant/rating scope')
46        if full.get('initialFen') not in ('startpos',chess.STARTING_FEN):raise ValueError('nonstandard start')
47        white,black=identity(full['white']),identity(full['black'])
48        if {white,black}!={bot.lower(),opponent.lower()}:raise ValueError('unauthorized player')
49        clock=full.get('clock',{})
50        if clock.get('initial',0)<600000 or clock.get('increment',0)<5000:raise ValueError('clock too fast')
51        self.id=full['id']
52        if not re.fullmatch(r'[A-Za-z0-9]{8}',self.id):raise ValueError('invalid game id')
53        self.color=white==bot.lower();self.full=full
54        self.directory=Path(directory)/self.id;self.directory.mkdir(parents=True,exist_ok=True)
55        path=self.directory/'journal.json'
56        self.journal=json.loads(path.read_text()) if path.exists() else {'decisions':{},'events':[]}
57        self.state=full['state'];self.board=reconstruct(self.state['moves'])
58        self.save()
59    def save(self):
60        self.journal.update(game_id=self.id,state=self.state,final_fen=self.board.fen())
61        atomic_json(self.directory/'journal.json',self.journal)
62        g=chess.pgn.Game();node=g;b=chess.Board()
63        for move in self.board.move_stack:
64            key=position_key(b);entry=self.journal['decisions'].get(key)
65            b.push(move);node=node.add_variation(move)
66            if entry and entry['uci']==move.uci():
67                meta=entry['metrics'];source=meta['decision_source']
68                if source=='forced_mate_override':
69                    node.comment=f"THREE-PLY MATE HELPER: proved mate within {meta['mate_search']['mate_plies']} plies; model not queried. Proof in local journal."
70                elif source=='legal_fallback':node.comment='Legal fallback: '+meta['reason']+'; model probabilities unavailable.'
71        winner=self.state.get('winner');status=self.state.get('status','started')
72        result='1-0' if winner=='white' else '0-1' if winner=='black' else '1/2-1/2' if status in ('draw','stalemate') else '*'
73        g.headers.update(Event='Lichess bot test: model + legality filter + three-ply mate helper',
74            Site='https://lichess.org/'+self.id,White=self.full['white'].get('name',identity(self.full['white'])),
75            Black=self.full['black'].get('name',identity(self.full['black'])),Result=result,Termination=status)
76        tmp=self.directory/'game.pgn.tmp';tmp.write_text(str(g)+'\n');tmp.replace(self.directory/'game.pgn')
77    def ingest(self,state):
78        self.board=reconstruct(state['moves']);self.state=state;self.save()
79    def decide(self,policy,fresh=False,age=0,now=None):
80        now=time.time() if now is None else now
81        if self.state['status']!='started' or self.board.turn!=self.color or self.board.is_game_over(claim_draw=False):return None
82        key=position_key(self.board);entry=self.journal['decisions'].get(key)
83        if entry:
84            # Accepted HTTP submission awaits server confirmation; never duplicate.
85            if entry['submission']=='accepted':return None
86            # A timeout may have committed the move. Only a NEW full snapshot
87            # confirming this exact unchanged history permits the same-move retry.
88            if not fresh or now-entry.get('last_attempt',now)<2 or entry.get('attempts',0)>=3:return None
89            return entry
90        started=time.monotonic();move,metrics=policy(self.board.copy(stack=True),budget(self.state,self.color,age))
91        if move not in self.board.legal_moves:raise ValueError('policy output illegal')
92        entry={'key':key,'ply':self.board.ply()+1,'uci':move.uci(),'metrics':metrics,
93            'decision_seconds':time.monotonic()-started,'attempts':0,'submission':'prepared'}
94        self.journal['decisions'][key]=entry;self.save();return entry
95    def mark(self,entry,status,now=None):
96        entry['submission']=status
97        if status=='sending':entry['attempts']+=1;entry['last_attempt']=time.time() if now is None else now
98        self.save()
99