app_multi.py

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