rust_mate.py

1"""Isolated Rust mate proof adapter. Full root/history, bounded subprocess, verified certificate."""
2import json,math,subprocess,time
3from pathlib import Path
4import chess
5from mate_ipc import invoke
6from mate_cpu_pool import acquire,release
7BINARY=Path(__file__).with_name('littleply-mate')
8if not BINARY.exists():BINARY=Path(__file__).parent/'target/release/littleply-mate'
9
10def response_schema(value,depth,nodes,seconds,workers):
11    if type(value) is not dict:raise ValueError('response object required')
12    def integer(name,lo,hi):
13        v=value[name]
14        if type(v) is not int or not lo<=v<=hi:raise ValueError(name)
15    for name,lo,hi in [('max_plies',depth,depth),('node_budget',nodes,nodes),('nodes',0,nodes),
16                       ('completed_depth_plies',0,depth),('deepest_visited_ply',0,depth),('workers',workers,workers)]:integer(name,lo,hi)
17    for name in ('seconds','time_budget_seconds'):
18        v=value[name]
19        if type(v) not in (int,float) or not math.isfinite(v) or not 0<=v<=60:raise ValueError(name)
20    if not 0<value['time_budget_seconds']<=seconds or value['seconds']>seconds+1:raise ValueError('timing range')
21    status=value['status'];reason=value['reason']
22    allowed={'proven':{'forced_checkmate'},'no_forced_mate':{'exhausted_depth_bound'},'unknown':{'deadline','node_budget'}}
23    if type(status) is not str or type(reason) is not str or status not in allowed or reason not in allowed[status]:raise ValueError('status')
24    if value['backend']!='rust-shakmaty-0.27.3':raise ValueError('backend')
25    iterations=value['iteration_depths']
26    expected=list(range(1,depth+1,2))
27    if type(iterations) is not list or not iterations or any(type(d) is not int for d in iterations) or iterations!=expected[:len(iterations)]:raise ValueError('iterations')
28    if status=='proven':
29        integer('mate_plies',1,depth)
30        if type(value['move']) is not str or len(value['move']) not in (4,5) or type(value['proof']) is not dict:raise ValueError('proof schema')
31    elif any(value[k] is not None for k in ('move','proof','mate_plies')):raise ValueError('unexpected proof')
32    if status=='no_forced_mate' and value['completed_depth_plies']!=depth:raise ValueError('incomplete disproof')
33    fields=('status','reason','max_plies','node_budget','nodes','completed_depth_plies','deepest_visited_ply','seconds','time_budget_seconds','backend','move','proof','mate_plies','iteration_depths','workers')
34    return {k:value[k] for k in fields}
35
36def forced_mate(board,*,adjudicate,max_plies=3,max_nodes=50000,seconds=1.0,workers=1,binary=BINARY):
37    if type(max_plies) is not int or not 1<=max_plies<=10:raise ValueError('depth is 1..10 PLIES')
38    if type(max_nodes) is not int or not 1<=max_nodes<=1000000 or not math.isfinite(seconds) or not 0<seconds<=10:raise ValueError('invalid bounded budget')
39    if type(board) is not chess.Board or board.chess960 or not board.is_valid():raise ValueError('valid standard chess required')
40    if type(workers) is not int or not 1<=workers<=8:raise ValueError('workers must be 1..8')
41    started=time.monotonic();deadline=started+seconds
42    permits=[]
43    def unknown(reason,raw=None):
44        result=dict(nodes=None,completed_depth_plies=0,deepest_visited_ply=None,proof_verified=False,proof_nodes=0,iteration_depths=[])
45        if raw is not None:
46            for k in ('nodes','completed_depth_plies','deepest_visited_ply','iteration_depths'):result[k]=raw[k]
47        result.update(status='unknown',reason=reason,move=None,proof=None,mate_plies=None,
48            backend='rust-shakmaty-0.27.3',max_plies=max_plies,node_budget=max_nodes,time_budget_seconds=seconds,seconds=time.monotonic()-started,workers=len(permits),requested_workers=workers)
49        return result
50    if seconds<.005:return unknown('insufficient_process_budget')
51    request=dict(root_fen=board.root().fen(),moves=[m.uci() for m in board.move_stack],max_plies=max_plies,max_nodes=max_nodes,millis=max(1,int(seconds*700)))
52    try:
53        permits=acquire(workers)
54        if not permits:return unknown('cpu_capacity')
55        request['workers']=len(permits)
56        try:code,out=invoke(binary,json.dumps(request),deadline,pass_fds=tuple(permits))
57        except TimeoutError:return unknown('process_deadline')
58        except OverflowError:return unknown('output_limit')
59        if code:return unknown('backend_error')
60        try:result=response_schema(json.loads(out),max_plies,max_nodes,seconds,len(permits))
61        except (ValueError,TypeError,KeyError,RecursionError,OverflowError):return unknown('invalid_response')
62        if time.monotonic()>=deadline:return unknown('process_deadline',result)
63        if result.get('status')!='proven':
64            if result['status']=='unknown':return unknown(result['reason'],result)
65            result['seconds']=time.monotonic()-started;result['time_budget_seconds']=seconds;return result
66        work=board.copy(stack=True);attacker=board.turn;proof_nodes=0
67        def verify(p,left):
68            nonlocal proof_nodes
69            if time.monotonic()>=deadline:raise TimeoutError
70            proof_nodes+=1
71            if proof_nodes>20000:raise TimeoutError
72            if type(p) is not dict or type(p.get('mate_plies')) is not int or type(p.get('branches')) is not list:raise ValueError('certificate schema')
73            if p['fen']!=work.fen():raise ValueError('certificate position')
74            end=adjudicate(work)
75            if p['kind']=='checkmate':
76                if not end or end[1]!='checkmate' or work.turn==attacker or p['mate_plies']!=0 or p['branches']:raise ValueError('certificate leaf')
77                return 0
78            if end or left<=0:raise ValueError('certificate terminal or depth')
79            branches=p['branches'];ucis=[b['move'] for b in branches]
80            expected='attacker_choice' if work.turn==attacker else 'all_defenses'
81            if p['kind']!=expected:raise ValueError('certificate role')
82            if work.turn==attacker:
83                if len(ucis)!=1:raise ValueError('certificate attacker choice')
84            elif sorted(ucis)!=sorted(m.uci() for m in work.legal_moves):raise ValueError('missing defense')
85            distances=[]
86            for b in branches:
87                m=chess.Move.from_uci(b['move'])
88                if m not in work.legal_moves:raise ValueError('illegal proof move')
89                work.push(m)
90                try:distances.append(verify(b['child'],left-1))
91                finally:work.pop()
92            distance=1+max(distances)
93            if p['mate_plies']!=distance:raise ValueError('certificate distance')
94            return distance
95        try:
96            distance=verify(result['proof'],max_plies)
97            if result['mate_plies']!=distance or result['move']!=result['proof']['branches'][0]['move']:raise ValueError('certificate root')
98            if time.monotonic()>=deadline:raise TimeoutError
99        except TimeoutError:return unknown('proof_verification_budget',result)
100        except (ValueError,KeyError,TypeError,IndexError,AttributeError,RecursionError):return unknown('invalid_proof',result)
101        result.update(proof_verified=True,proof_nodes=proof_nodes,seconds=time.monotonic()-started,time_budget_seconds=seconds)
102        return result
103    except (OSError,ValueError,TypeError):return unknown('backend_unavailable_or_invalid')
104    finally:release(permits)
105