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