app_multi.py

1"""Candidate multi-game controller. Not deployed until independent review."""
2import argparse,collections,os,queue,re,time
3from dataclasses import dataclass
4from pathlib import Path
5import chess
6from app import Service as SingleService,fresh_history,notify
7from core import identity,position_key,atomic_json
8from transport import Stream
9from multi_game import RealtimeGame
10from scheduler import Scheduler,Request,clock_window,realtime_reason,expected_policy_seconds,bullet_clock
11
12@dataclass
13class Slot:
14    gid:str
15    stream:object
16    game:object=None
17    epoch:int=0
18    received:float=0.
19    pending:object=None
20    inflight:object=None
21
22def admission(c,bot,busy=False):
23    if identity(c.get('destUser',{}))!=bot.lower() or not identity(c.get('challenger',{})):return 'generic'
24    if c.get('rated') is not False:return 'casual'
25    if c.get('variant',{}).get('key')!='standard' or c.get('initialFen') not in (None,'startpos',chess.STARTING_FEN):return 'standard'
26    reason=realtime_reason(c.get('timeControl',{}))
27    return reason or ('later' if busy else None)
28
29class Service(SingleService):
30    def __init__(self,args):
31        super().__init__(args)
32        self.cap=int(self.manifest.get('max_games',4));assert 1<=self.cap<=4
33        self.slots={};self.reservations={};self.retired=[];self.scheduler=Scheduler(10)
34        self.owner.poll_floor=8 # soft clock expiry never kills a useful shared worker
35        self.samples=collections.deque(maxlen=64);self.last_snapshot=0.
36    def release_reserved(self,cid):self.reservations.pop(cid,None)
37    def accept_capacity_valid(self,cid):
38        count=len(self.slots)+len(self.reservations)
39        if cid not in self.reservations or count>self.cap:return False
40        bullet=any(r.get('bullet',False) for r in self.reservations.values())
41        for slot in self.slots.values():
42            if not slot.game:return False
43            clock=slot.game.full['clock']
44            bullet=bullet or bullet_clock(clock['initial']/1000,clock['increment']/1000)
45        return not bullet or count==1
46    def start_game(self,gid):
47        if not re.fullmatch('[A-Za-z0-9]{8}',gid) or gid in self.closed or gid in self.slots:return
48        if len(self.slots)>=10:raise RuntimeError('unexpected account concurrency above recovery bound')
49        self.accepts().invalidate(gid)
50        self.reservations.pop(gid,None)
51        self.slots[gid]=Slot(gid,Stream(self.api,'/api/bot/game/stream/'+gid,gid,self.events,self.epochs))
52        self.event('game_stream_started',game=gid)
53    def identity_matches(self,gid,epoch,digest,release=None):
54        slot=self.slots.get(gid);g=slot.game if slot else None
55        return bool(g and slot.epoch==epoch and (release is None or release==self.version) and g.state['status']=='started' and position_key(g.board)==digest)
56    def valid(self,key):return self.identity_matches(*key)
57    def retry_local_slot(self,slot,entry,delay=0):
58        g=slot.game
59        if g and g.state['status']=='started' and g.board.turn==g.color and position_key(g.board)==entry['key'] and chess.Move.from_uci(entry['uci']) in g.board.legal_moves:
60            slot.pending=(slot.epoch,entry['key'],entry,time.monotonic()+delay)
61    def response(self,key,status,data):
62        if key.startswith('move:'):
63            gid=key.split(':')[1];slot=self.slots.get(gid)
64            if status=='send_ready':
65                try:
66                    job=self.jobs.get(key)
67                    if slot and slot.game and slot.inflight and slot.inflight[0]==key and job and time.monotonic()<data.expires and (not job[4] or job[4]()):
68                        slot.game.mark(slot.inflight[1],'sending');data.allowed=True
69                finally:data.event.set()
70                return
71            self.jobs.pop(key,None)
72            if status=='auth_error':raise RuntimeError('move authorization failed')
73            if not slot or not slot.game or not slot.inflight or slot.inflight[0]!=key:return
74            _,entry=slot.inflight;slot.inflight=None;g=slot.game
75            if status=='ok':g.mark(entry,'accepted');self.event('move_accepted',game=gid,ply=entry['ply'],uci=entry['uci'],source=entry['metrics']['decision_source'])
76            elif status=='cancelled':
77                if entry['submission']=='sending':entry['attempts']=max(0,entry['attempts']-1)
78                g.mark(entry,'local_pending');self.retry_local_slot(slot,entry)
79            else:
80                g.mark(entry,'uncertain' if status in ('uncertain','retryable') else 'rejected')
81                slot.stream.resync.set();self.event('move_resync',game=gid,status=status)
82            return
83        if key=='playing' and status=='ok':
84            job=self.jobs.pop(key,None);games=data.get('nowPlaying',[]);self.synced=True;self.last_snapshot=time.monotonic()
85            for game in games:self.start_game(game.get('gameId',''))
86            self.reconcile_accepts(games,job)
87            return
88        super().response(key,status,data)
89    def windows(self,challenge=None):
90        now=time.monotonic();windows=[];costs=[]
91        for slot in self.slots.values():
92            g=slot.game
93            if g:
94                age=max(0,now-slot.received) if g.board.turn==g.color else 0
95                windows.append(clock_window(g.state,g.color,age));costs.append(expected_policy_seconds(1+g.board.ply()*3))
96            else:windows.append(1.);costs.append(.61)
97        for res in self.reservations.values():windows.append(res['window']);costs.append(.61)
98        if challenge:
99            tc=challenge['timeControl'];remaining=max(tc['limit'],tc['increment'],1)
100            windows.append(clock_window({'wtime':remaining*1000,'winc':tc['increment']*1000},True));costs.append(.61)
101        scale=max(1.,max(self.samples,default=0)*1.1)
102        return windows,[cost*scale for cost in costs]
103    def accepting(self,challenge=None):
104        healthy=bool(self.auth and self.synced and self.online and self.account_live and self.owner.ready and not self.bad_policy
105            and not self.drain and not (self.state/'drain').exists() and self.api.cooldown.remaining()==0
106            and len(self.slots)+len(self.reservations)<self.cap and len(self.retired)<self.cap
107            and __import__('shutil').disk_usage(self.state).free>512*1024**2)
108        if not healthy:return False
109        # The measured CPU burst is unsuitable for multiple simultaneous bullet
110        # clocks. All Bot-API-allowed bullet settings remain supported when idle.
111        if challenge:
112            tc=challenge['timeControl']
113            if bullet_clock(tc['limit'],tc['increment']) and (self.slots or self.reservations):return False
114        for slot in self.slots.values():
115            if not slot.game:return False # wait for authoritative clock metadata
116            clock=slot.game.full['clock']
117            if bullet_clock(clock['initial']/1000,clock['increment']/1000):return False
118        if any(r.get('bullet',False) for r in self.reservations.values()):return False
119        windows,costs=self.windows(challenge)
120        return not windows or sum(costs)<=min(windows)
121    def ingest(self,tag,kind,data,epoch,received):
122        if kind=='auth_error':raise RuntimeError('stream authentication failed')
123        if tag=='account':
124            if kind=='connected':self.account_live=True;self.reconcile();self.api_job('online','GET','/api/users/status?ids='+self.args.bot)
125            elif kind=='disconnected':self.account_live=False;self.event('account_disconnected',reason=data)
126            elif kind=='data':
127                typ=data.get('type')
128                if typ=='challenge':
129                    c=data['challenge'];cid=c.get('id','')
130                    if not re.fullmatch('[A-Za-z0-9]{8}',cid) or cid in self.reservations:return
131                    reason=admission(c,self.args.bot)
132                    if not reason and not self.accepting(c):reason='later'
133                    if reason:
134                        self.api_job('decline:'+cid,'POST',f'/api/challenge/{cid}/decline',{'reason':reason},priority=2)
135                        self.event('challenge_declined',id=cid,reason=reason,clock=c.get('timeControl'),capacity=len(self.slots)+len(self.reservations))
136                    else:
137                        windows,_=self.windows(c);tc=c['timeControl']
138                        self.reservations[cid]={'time':time.monotonic(),'window':windows[-1],'bullet':bullet_clock(tc['limit'],tc['increment'])}
139                        self.begin_accept(cid)
140                elif typ=='gameStart':self.start_game(data['game'].get('gameId') or data['game']['id'])
141                elif typ in ('challengeCanceled','challengeDeclined'):self.cancel_accept(data.get('challenge',{}).get('id'))
142            return
143        slot=self.slots.get(tag)
144        if not slot or kind!='data':return
145        typ=data.get('type')
146        if typ=='gameFull':
147            if epoch<slot.epoch:return
148            slot.epoch=epoch
149            if slot.game:
150                if not fresh_history(slot.game.state['moves'],data['state']['moves']):self.event('stale_snapshot_ignored',game=tag);return
151                slot.game.ingest(data['state'])
152            else:
153                slot.game=RealtimeGame(data,self.args.bot,self.state/'games')
154                pinned=slot.game.journal.get('release')
155                if pinned and pinned!=self.version:raise RuntimeError('active game pinned to different release')
156                slot.game.journal['release']=self.version;slot.game.save();self.event('game_started',game=tag,release=self.version)
157            fresh=True
158        elif typ=='gameState' and slot.game and epoch==slot.epoch:
159            if not fresh_history(slot.game.state['moves'],data['moves']):self.event('stale_state_ignored',game=tag);return
160            slot.game.ingest(data);fresh=False
161        else:return
162        slot.received=received;g=slot.game
163        if g.state['status']!='started':
164            self.event('game_finished',game=tag,status=g.state['status'],winner=g.state.get('winner'))
165            self.closed.add(tag)
166            if len(self.closed)>256:self.closed.pop()
167            slot.stream.stop.set();self.retired.append(slot.stream);self.slots.pop(tag);self.reservations.pop(tag,None);self.scheduler.remove(tag)
168            return
169        waiting=self.scheduler.waiting.get(tag)
170        if waiting and not self.valid(waiting.key):self.scheduler.remove(tag)
171        if fresh and not slot.inflight:
172            entry=g.journal['decisions'].get(position_key(g.board))
173            if entry and entry['submission']!='accepted' and entry.get('attempts',0)<3:self.retry_local_slot(slot,entry,max(2,self.api.cooldown.remaining()))
174    def decision(self,slot,request,status,answer):
175        if not self.valid(request.key):self.event('stale_inference_discarded',game=request.key[0]);return
176        g=slot.game;digest=request.key[2]
177        if digest in g.journal['decisions']:return
178        if status=='ok':
179            uci,metrics=answer
180            if chess.Move.from_uci(uci) not in g.board.legal_moves:raise ValueError('illegal inference output')
181            self.bad_policy=False
182        else:
183            uci=sorted(g.board.legal_moves,key=lambda m:m.uci())[0].uci()
184            query='not_queried' if status=='context' or 'model not queried' in str(answer) else 'in_progress_at_deadline' if status=='deadline' else 'unavailable'
185            metrics={'decision_source':'legal_fallback','model_query':query,'reason':str(answer)}
186            self.event('explicit_legal_fallback',game=g.id,reason=str(answer));self.bad_policy=True
187        metrics['queue_seconds']=max(0,getattr(request,'dispatched',time.monotonic())-request.arrived)
188        entry={'key':digest,'ply':g.board.ply()+1,'uci':uci,'metrics':metrics,'attempts':0,'submission':'prepared'}
189        g.journal['decisions'][digest]=entry;g.save();self.retry_local_slot(slot,entry)
190    def work(self):
191        try:
192            while True:
193                key,status,answer=self.owner.results.get_nowait();request=self.scheduler.complete(key)
194                slot=self.slots.get(key[0])
195                if request and status=='ok':
196                    self.samples.append(answer[1].get('decision_seconds',0)/expected_policy_seconds(1+request.board.ply()*3))
197                if request and slot and not request.expired:self.decision(slot,request,status,answer)
198                else:self.event('late_inference_discarded',game=key[0])
199        except queue.Empty:pass
200        now=time.monotonic();running=self.scheduler.running
201        if running and not running.expired and now>=running.deadline:
202            running.expired=True;slot=self.slots.get(running.key[0])
203            if slot:self.decision(slot,running,'deadline','clock deadline while inference finishes; no worker interruption')
204        for slot in list(self.slots.values()):
205            g=slot.game
206            if not g:continue
207            if slot.pending and not slot.inflight and now>=slot.pending[3] and not self.api.cooldown.remaining():
208                epoch,digest,entry,_=slot.pending;slot.pending=None
209                if self.identity_matches(slot.gid,epoch,digest):
210                    key='move:'+slot.gid+':'+digest;gid=slot.gid
211                    guard=lambda gid=gid,epoch=epoch,digest=digest:self.identity_matches(gid,epoch,digest)
212                    g.mark(entry,'queued');slot.inflight=(key,entry)
213                    if not self.api_job(key,'POST',f'/api/bot/game/{gid}/move/{entry["uci"]}',priority=0,guard=guard):
214                        slot.inflight=None;g.mark(entry,'local_pending');self.retry_local_slot(slot,entry,.2)
215            digest=position_key(g.board);key=(slot.gid,slot.epoch,digest,self.version)
216            running_same=self.scheduler.running and self.scheduler.running.key==key
217            if not slot.pending and not slot.inflight and not running_same and g.state['status']=='started' and g.board.turn==g.color and digest not in g.journal['decisions'] and not g.board.is_game_over():
218                if 1+g.board.ply()*3+2>1024:
219                    request=Request(key,g.board.copy(stack=True),now,now,0)
220                    self.decision(slot,request,'context','full history exceeds context; no truncation; model not queried')
221                else:self.scheduler.add(key,g.board.copy(stack=True),clock_window(g.state,g.color,max(0,now-slot.received)),now)
222        # Waiting games have their own clocks even while another request owns
223        # the CPU. Expire them without waiting for that computation to finish.
224        for gid,request in list(self.scheduler.waiting.items()):
225            if time.monotonic()>=request.deadline:
226                self.scheduler.remove(gid);slot=self.slots.get(gid)
227                if slot:self.decision(slot,request,'deadline','clock deadline in inference queue; model not queried')
228        if self.owner.ready and not self.owner.busy:
229            request=self.scheduler.next(self.valid,now)
230            if request:
231                left=request.deadline-time.monotonic();slot=self.slots.get(request.key[0])
232                if left<=.05:
233                    self.scheduler.complete(request.key)
234                    if slot:self.decision(slot,request,'deadline','clock deadline in inference queue; model not queried')
235                else:
236                    request.dispatched=time.monotonic();self.owner.submit(request.key,request.board,min(8,left))
237    def run(self):
238        try:
239            while True:
240                for _ in range(256):
241                    try:self.ingest(*self.events.get_nowait())
242                    except queue.Empty:break
243                for _ in range(64):
244                    try:self.response(*self.api.results.get_nowait())
245                    except queue.Empty:break
246                now=time.monotonic();self.retired=[s for s in self.retired if s.thread.is_alive()]
247                for task in self.retry[:]:
248                    due,key,job=task
249                    if now>=due:self.retry.remove(task);self.api_job(key,*job[:3],priority=job[3],guard=job[4])
250                if self.reservations and now-self.last_reconcile>10:self.reconcile()
251                if self.auth and self.account_live and not self.online and now-self.last_online_check>10:
252                    self.last_online_check=now;self.api_job('online','GET','/api/users/status?ids='+self.args.bot)
253                self.work()
254                if self.bad_policy and self.owner.ready and not self.slots:self.bad_policy=False
255                if now-self.last_health>=2:
256                    self.last_health=now;ready=self.auth and self.synced and self.account_live and self.online and self.owner.ready
257                    games=list(self.slots)
258                    atomic_json(self.state/'health.json',dict(updated=time.time(),ready=bool(ready),accepting=self.accepting(),active_games=games,
259                        active_game=games[0] if len(games)==1 else None,release=self.version,device=__import__('inference').serving_device(),max_games=self.cap,
260                        waiting=len(self.scheduler.waiting),worker_pid=self.owner.p.pid if self.owner.p else None,launcher_pid=os.getpid(),
261                        cooldown_seconds=self.api.cooldown.remaining(),draining=self.drain or (self.state/'drain').exists()))
262                    if ready and not self.notified:notify('READY=1');self.notified=True;self.event('service_ready',public=True,max_games=self.cap)
263                    notify('WATCHDOG=1')
264                if (self.drain or (self.state/'drain').exists()) and not self.slots and not self.reservations:break
265                time.sleep(.01)
266        finally:
267            self.drain=True
268            for slot in self.slots.values():slot.stream.close()
269            for stream in self.retired:stream.close()
270            if self.account:self.account.close()
271            self.api.close();self.owner.close();self.event('service_stopped');self.lock.close()
272            atomic_json(self.state/'health.json',dict(updated=time.time(),ready=False,accepting=False,stopped=True))
273
274def main():
275    p=argparse.ArgumentParser();p.add_argument('--release',type=Path,required=True);p.add_argument('--state',type=Path,required=True)
276    p.add_argument('--token-file',type=Path,required=True);p.add_argument('--bot',default='LittlePly');Service(p.parse_args()).run()
277