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