app.py

1"""Conservative single-game public Bot API service; no matchmaking or chat."""
2import argparse,fcntl,hashlib,itertools,json,logging,logging.handlers,os,queue,re,signal,socket,sys,time
3from pathlib import Path
4from core import Game,identity,position_key,atomic_json,budget
5from transport import Transport,Stream,Events
6from inference import Owner
7from acceptance import Acceptances
8import chess
9
10def admission(c,bot,busy=False):
11    if busy:return 'later'
12    if identity(c.get('destUser',{}))!=bot.lower():return 'generic'
13    if not identity(c.get('challenger',{})):return 'generic'
14    if c.get('rated') is not False:return 'casual'
15    if c.get('variant',{}).get('key')!='standard' or c.get('initialFen') not in (None,'startpos',chess.STARTING_FEN):return 'standard'
16    tc=c.get('timeControl',{})
17    if tc.get('type')!='clock' or not 600<=tc.get('limit',0)<=1800 or not 5<=tc.get('increment',0)<=30:return 'timeControl'
18    return None
19
20def fresh_history(old,new):
21    a=old.split();b=new.split()
22    return len(b)>=len(a) and b[:len(a)]==a
23
24def notify(message):
25    address=os.environ.get('NOTIFY_SOCKET')
26    if address:
27        if address.startswith('@'):address='\0'+address[1:]
28        with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as s:s.connect(address);s.sendall(message.encode())
29
30def verify_release(root):
31    manifest=json.loads((root/'manifest.json').read_text())
32    if manifest.get('identity_version')==2:
33        identity={k:manifest[k] for k in ('files','entrypoint','max_games')}
34        expected=manifest['checkpoint_sha256'][:12]+'-'+hashlib.sha256(json.dumps(identity,sort_keys=True).encode()).hexdigest()[:12]
35        if manifest['release_id']!=expected or root.resolve().name!=expected:raise ValueError('release configuration identity mismatch')
36        if manifest['files'].get('checkpoint.pt')!=manifest['checkpoint_sha256']:raise ValueError('checkpoint identity mismatch')
37    for name,expected in manifest['files'].items():
38        p=(root/name).resolve()
39        if root.resolve() not in p.parents:raise ValueError('invalid manifest path')
40        h=hashlib.sha256()
41        with p.open('rb') as f:
42            for block in iter(lambda:f.read(1048576),b''):h.update(block)
43        if h.hexdigest()!=expected:raise ValueError('release hash mismatch: '+name)
44    return manifest
45
46class Service:
47    def __init__(self,args):
48        self.args=args;self.root=args.release.resolve();self.manifest=verify_release(self.root)
49        self.version=self.manifest['release_id'];self.state=args.state;self.state.mkdir(parents=True,exist_ok=True)
50        self.lock=(self.state/'account.lock').open('a');fcntl.flock(self.lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
51        self.log=logging.getLogger('littleply');self.log.setLevel(logging.INFO)
52        handler=logging.handlers.RotatingFileHandler(self.state/'events.jsonl',maxBytes=5_000_000,backupCount=3)
53        self.log.addHandler(handler)
54        self.api=Transport(args.token_file);self.owner=Owner(self.root/'checkpoint.pt')
55        self.events=Events();self.epochs=itertools.count(1);self.account=None;self.stream=None
56        self.game=None;self.epoch=0;self.pending=None;self.inflight=None;self.reservation=None;self.drain=False
57        self.auth=False;self.synced=False;self.online=False;self.account_live=False;self.last_health=0;self.notified=False
58        self.jobs={};self.retry=[];self.last_reconcile=0.;self.last_online_check=0.;self.cool_until=0.;self.bad_policy=False;self.closed=set()
59        self.api_job('account','GET','/api/account')
60        signal.signal(signal.SIGTERM,lambda *_:self.shutdown());signal.signal(signal.SIGINT,lambda *_:self.shutdown())
61    def event(self,event,**data):self.log.info(json.dumps(dict(utc=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),event=event,**data)))
62    def shutdown(self):self.drain=True;self.event('draining')
63    def api_job(self,key,method,path,data=None,priority=1,guard=None):
64        if key in self.jobs:return False
65        self.jobs[key]=(method,path,data,priority,guard,time.monotonic())
66        try:self.api.submit(key,method,path,data,priority,guard)
67        except queue.Full:self.jobs.pop(key);self.event('control_queue_full');return False
68        return True
69    def accepts(self):
70        if not hasattr(self,'acceptances'):self.acceptances=Acceptances()
71        return self.acceptances
72    def release_reserved(self,cid):
73        if self.reservation and self.reservation[0]==cid and not self.stream:self.reservation=None
74    def accept_capacity_valid(self,cid):
75        return bool(self.reservation and self.reservation[0]==cid and not self.stream and not self.game)
76    def begin_accept(self,cid):
77        op=self.accepts().begin(cid)
78        guard=lambda:self.accepts().valid(op.key)
79        if not self.api_job(op.key,'POST',f'/api/challenge/{cid}/accept',guard=guard):
80            self.accepts().invalidate(cid);self.release_reserved(cid)
81    def cancel_accept(self,cid):
82        if self.accepts().cancel(cid):self.release_reserved(cid)
83    def reconcile_accepts(self,games,job):
84        for cid in self.accepts().reconcile({g.get('gameId') for g in games},job[5] if job and len(job)>5 else 0,self.account_live):
85            self.release_reserved(cid);self.event('accept_reconciled_absent',id=cid)
86    def reconcile(self):
87        if 'playing' not in self.jobs:self.api_job('playing','GET','/api/account/playing',priority=0);self.last_reconcile=time.monotonic()
88    def start_game(self,gid):
89        if not re.fullmatch(r'[A-Za-z0-9]{8}',gid):return
90        if gid in self.closed:return
91        if self.game and self.game.id==gid:return
92        if self.stream:
93            if self.stream.tag==gid:return
94            self.drain=True;self.event('unexpected_second_game',game=gid);return
95        self.accepts().invalidate(gid)
96        self.reservation=(gid,time.monotonic());self.stream=Stream(self.api,'/api/bot/game/stream/'+gid,gid,self.events,self.epochs)
97        self.event('game_stream_started',game=gid)
98    def response(self,key,status,data):
99        if status=='send_ready':
100            # Queue insertion is not a network attempt. Persist uncertainty only
101            # when the network owner is ready and this exact request is current.
102            try:
103                job=self.jobs.get(key)
104                if key.startswith('accept:'):
105                    op=self.accepts().find(key)
106                    if op and job and time.monotonic()<data.expires and job[4]() and self.account_live and not self.drain and not (self.state/'drain').exists() and self.accept_capacity_valid(op.cid) and self.accepts().sent(key):data.allowed=True
107                    return
108                if self.game and self.inflight and key==self.inflight[0] and job and time.monotonic()<data.expires:
109                    guard=job[4]
110                    if not guard or guard():
111                        self.game.mark(self.inflight[1],'sending');data.allowed=True
112            finally:data.event.set()
113            return
114        job=self.jobs.pop(key,None)
115        if status=='auth_error':self.event('authentication_failed');raise RuntimeError('authentication unavailable')
116        if key.startswith('move:'):
117            if not self.game or not self.inflight or key!=self.inflight[0]:return
118            _,entry=self.inflight;self.inflight=None
119            if status=='ok':self.game.mark(entry,'accepted');self.event('move_accepted',game=self.game.id,ply=entry['ply'],uci=entry['uci'],source=entry['metrics']['decision_source'])
120            elif status=='cancelled':
121                # Transport guarantees that cancelled requests were never sent.
122                if entry['submission']=='sending':entry['attempts']=max(0,entry['attempts']-1)
123                self.game.mark(entry,'local_pending')
124                self.retry_local(entry)
125            else:
126                self.game.mark(entry,'uncertain' if status in ('uncertain','retryable') else 'rejected')
127                self.event('move_resync',game=self.game.id,status=status)
128                if self.stream:self.stream.resync.set()
129            return
130        if key.startswith('accept:'):
131            op=self.accepts().find(key)
132            if not op:return
133            cid=self.accepts().settle(key,status)
134            if cid:self.release_reserved(cid)
135            self.event('accept_result',id=op.cid,status=status)
136            self.reconcile()
137            return
138        if status in ('rate_limited','retryable'):
139            self.event('control_retry',operation=key,status=status)
140            if job and not key.startswith('decline:'):self.retry.append((time.monotonic()+max(5,self.api.cooldown.remaining()),key,job))
141            return
142        if status!='ok':self.event('control_result',operation=key,status=status);return
143        if key=='account':
144            if data.get('id')!=self.args.bot.lower() or data.get('title')!='BOT':raise RuntimeError('BOT identity mismatch')
145            self.auth=True;self.account=Stream(self.api,'/api/stream/event','account',self.events,self.epochs);self.reconcile()
146            self.api_job('online','GET','/api/users/status?ids='+self.args.bot)
147        elif key=='online':self.online=bool(data and data[0].get('online'))
148        elif key=='playing':
149            games=data.get('nowPlaying',[]);self.synced=True
150            if len(games)>1:self.drain=True;self.event('unexpected_multiple_active_games',count=len(games))
151            for g in games[:1]:self.start_game(g.get('gameId',''))
152            self.reconcile_accepts(games,job)
153    def ingest(self,tag,kind,data,epoch,received):
154        if kind=='auth_error':raise RuntimeError('stream authentication unavailable')
155        if tag=='account':
156            if kind=='connected':self.account_live=True;self.reconcile();self.api_job('online','GET','/api/users/status?ids='+self.args.bot)
157            elif kind=='disconnected':self.account_live=False;self.event('account_disconnected',reason=data)
158            elif kind=='data':
159                typ=data.get('type')
160                if typ=='challenge':
161                    c=data['challenge'];cid=c.get('id','')
162                    if not re.fullmatch(r'[A-Za-z0-9]{8}',cid):return
163                    if self.reservation and self.reservation[0]==cid:return
164                    reason=admission(c,self.args.bot,not self.accepting())
165                    if reason:
166                        self.api_job('decline:'+cid,'POST',f'/api/challenge/{cid}/decline',{'reason':reason},priority=2)
167                        self.event('challenge_declined',id=cid,reason=reason)
168                    else:
169                        self.reservation=(cid,time.monotonic())
170                        self.begin_accept(cid)
171                elif typ=='gameStart':self.start_game(data['game'].get('gameId') or data['game']['id'])
172                elif typ in ('challengeCanceled','challengeDeclined'):
173                    self.cancel_accept(data.get('challenge',{}).get('id'))
174            return
175        if not self.stream or tag!=self.stream.tag or kind!='data':return
176        typ=data.get('type')
177        if typ=='gameFull':
178            if epoch<self.epoch:return
179            self.epoch=epoch
180            if self.game:
181                if not fresh_history(self.game.state['moves'],data['state']['moves']):self.event('stale_snapshot_ignored');return
182                self.game.ingest(data['state'])
183            else:
184                players=[identity(data.get(c,{})) for c in ('white','black')]
185                if self.args.bot.lower() not in players:raise ValueError('wrong bot game')
186                other=players[1] if players[0]==self.args.bot.lower() else players[0]
187                self.game=Game(data,self.args.bot,other,self.state/'games')
188                pinned=self.game.journal.get('release')
189                if pinned and pinned!=self.version:raise RuntimeError('active game belongs to another release; restart pinned release')
190                self.game.journal['release']=self.version;self.game.save();self.event('game_started',game=self.game.id,release=self.version)
191            fresh=True
192        elif typ=='gameState' and self.game and epoch==self.epoch:
193            if not fresh_history(self.game.state['moves'],data['moves']):self.event('stale_state_ignored');return
194            self.game.ingest(data);fresh=False
195        else:return
196        self.state_received=received
197        if self.game.state['status']!='started':
198            self.event('game_finished',game=self.game.id,status=self.game.state['status'],winner=self.game.state.get('winner'))
199            self.closed.add(self.game.id)
200            if len(self.closed)>256:self.closed.pop()
201            self.stream.stop.set();self.stream=None;self.game=None;self.reservation=None;self.pending=None;self.inflight=None
202            self.cool_until=time.monotonic()+15;return
203        if fresh and not self.inflight:
204            entry=self.game.journal['decisions'].get(position_key(self.game.board))
205            if entry and entry['submission']!='accepted' and entry.get('attempts',0)<3:
206                self.pending=(self.epoch,position_key(self.game.board),entry,time.monotonic()+max(2,self.api.cooldown.remaining()))
207    def retry_local(self,entry,delay=0):
208        g=self.game
209        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:
210            self.pending=(self.epoch,entry['key'],entry,time.monotonic()+delay)
211    def accepting(self):
212        return bool(self.auth and self.synced and self.online and self.account_live and self.owner.ready and not self.bad_policy
213            and not self.drain and not (self.state/'drain').exists() and not self.game and not self.reservation
214            and time.monotonic()>=self.cool_until and self.api.cooldown.remaining()==0
215            and __import__('shutil').disk_usage(self.state).free>512*1024**2)
216    def identity_matches(self,gid,epoch,digest):
217        g=self.game
218        return bool(g and g.id==gid and self.epoch==epoch and g.state['status']=='started' and position_key(g.board)==digest)
219    def work(self):
220        try:
221            while True:
222                key,status,answer=self.owner.results.get_nowait()
223                gid,epoch,digest=key
224                if not self.identity_matches(gid,epoch,digest):self.event('stale_inference_discarded');continue
225                if status=='ok':
226                    uci,metrics=answer;move=chess.Move.from_uci(uci)
227                    if move not in self.game.board.legal_moves:raise ValueError('illegal policy output')
228                    self.bad_policy=False
229                else:
230                    self.bad_policy=True;uci=sorted(self.game.board.legal_moves,key=lambda m:m.uci())[0].uci()
231                    metrics={'decision_source':'legal_fallback','model_query':'unavailable','reason':'inference failure: '+str(answer)}
232                    self.event('explicit_legal_fallback',game=gid,reason=str(answer))
233                entry={'key':digest,'ply':self.game.board.ply()+1,'uci':uci,'metrics':metrics,'attempts':0,'submission':'prepared'}
234                self.game.journal['decisions'][digest]=entry;self.game.save();self.pending=(epoch,digest,entry,time.monotonic())
235        except queue.Empty:pass
236        if self.pending and not self.inflight and time.monotonic()>=self.pending[3] and not self.api.cooldown.remaining():
237            epoch,digest,entry,_=self.pending;self.pending=None;g=self.game
238            if g and self.identity_matches(g.id,epoch,digest):
239                key='move:'+g.id+':'+digest;gid=g.id
240                guard=lambda:self.identity_matches(gid,epoch,digest)
241                g.mark(entry,'queued');self.inflight=(key,entry)
242                if not self.api_job(key,'POST',f'/api/bot/game/{gid}/move/{entry["uci"]}',priority=0,guard=guard):
243                    self.inflight=None;g.mark(entry,'local_pending');self.retry_local(entry,.2)
244        g=self.game
245        if g and not self.pending and not self.inflight and not self.owner.busy and self.owner.ready:
246            digest=position_key(g.board)
247            if g.state['status']=='started' and g.board.turn==g.color and digest not in g.journal['decisions'] and not g.board.is_game_over():
248                self.owner.submit((g.id,self.epoch,digest),g.board.copy(stack=True),budget(g.state,g.color,age=max(0,time.monotonic()-self.state_received)))
249    def run(self):
250        try:
251            while True:
252                for _ in range(256):
253                    try:self.ingest(*self.events.get_nowait())
254                    except queue.Empty:break
255                for _ in range(64):
256                    try:self.response(*self.api.results.get_nowait())
257                    except queue.Empty:break
258                now=time.monotonic()
259                for task in self.retry[:]:
260                    due,key,job=task
261                    if now>=due:
262                        self.retry.remove(task)
263                        # Never replay an accept after uncertain reservation/state changed.
264                        if key.startswith('accept:'):self.reconcile()
265                        else:self.api_job(key,*job[:3],priority=job[3],guard=job[4])
266                if self.reservation and not self.game and now-self.last_reconcile>10:self.reconcile()
267                if self.auth and self.account_live and not self.online and now-self.last_online_check>10:
268                    self.last_online_check=now;self.api_job('online','GET','/api/users/status?ids='+self.args.bot)
269                self.work()
270                if self.bad_policy and self.owner.ready and not self.game:self.bad_policy=False
271                if now-self.last_health>=2:
272                    self.last_health=now
273                    ready=self.auth and self.synced and self.account_live and self.online and self.owner.ready
274                    atomic_json(self.state/'health.json',dict(updated=time.time(),ready=bool(ready),accepting=self.accepting(),active_game=self.game.id if self.game else None,
275                        release=self.version,device=__import__('inference').serving_device(),max_games=1,worker_pid=self.owner.p.pid if self.owner.p else None,
276                        launcher_pid=os.getpid(),cooldown_seconds=self.api.cooldown.remaining(),draining=self.drain or (self.state/'drain').exists()))
277                    if ready and not self.notified:notify('READY=1');self.notified=True;self.event('service_ready',public=True,max_games=1)
278                    notify('WATCHDOG=1')
279                if (self.drain or (self.state/'drain').exists()) and not self.game and not self.reservation:break
280                time.sleep(.05)
281        finally:
282            self.drain=True
283            if self.stream:self.stream.close()
284            if self.account:self.account.close()
285            self.api.close();self.owner.close();self.event('service_stopped');self.lock.close()
286            atomic_json(self.state/'health.json',dict(updated=time.time(),ready=False,accepting=False,stopped=True))
287
288def main():
289    p=argparse.ArgumentParser();p.add_argument('--release',type=Path,required=True);p.add_argument('--state',type=Path,required=True)
290    p.add_argument('--token-file',type=Path,required=True);p.add_argument('--bot',default='LittlePly');args=p.parse_args()
291    Service(args).run()
292