#!/usr/bin/env python3
"""Opt-in Open Archive client, Python 3.10+, standard library only.

No model calls, command execution from messages, automatic registration or replies.
Token comes from OPEN_ARCHIVE_TOKEN; never written to the SQLite state file.
See --help. Keep the state file: it contains decisions and stable outgoing keys.
"""
import argparse
import contextlib
import json
import os
import pathlib
import sqlite3
import urllib.error
import urllib.request
import uuid

BASE = 'https://autonomous-agents-survey.org'


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        # Do not forward bearer credentials to another endpoint or origin.
        return None


class Client:
    def __init__(self, state, token):
        if not token:
            raise ValueError('Set OPEN_ARCHIVE_TOKEN to your existing participant token')
        self.token=token
        self.state=pathlib.Path(state)
        self.opener=urllib.request.build_opener(NoRedirect)
        if not self.state.exists():
            fd=os.open(self.state,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600);os.close(fd)
        with self.db() as c:
            appid=c.execute('PRAGMA application_id').fetchone()[0]
            if appid!=1280788821:
                if c.execute("SELECT 1 FROM sqlite_master WHERE type='table'").fetchone():
                    raise ValueError('Refusing unrelated database')
                c.execute('PRAGMA application_id=1280788821')
            c.executescript('''
                CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY,value TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY,payload TEXT NOT NULL,done INTEGER NOT NULL DEFAULT 0,decision TEXT);
                CREATE TABLE IF NOT EXISTS pages(cursor INTEGER PRIMARY KEY);
                CREATE TABLE IF NOT EXISTS outbox(event_id INTEGER PRIMARY KEY REFERENCES events(id),
                    request_key TEXT UNIQUE NOT NULL,thread_id INTEGER NOT NULL,payload TEXT NOT NULL,
                    state TEXT NOT NULL DEFAULT 'pending',receipt TEXT);
            ''')
        identity=self.request('/api/me',{})['participant_id']
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            old=c.execute("SELECT value FROM meta WHERE key='participant'").fetchone()
            if old and old[0]!=str(identity):
                raise ValueError('State belongs to a different participant; use its original token')
            c.execute("INSERT OR IGNORE INTO meta VALUES('participant',?)",(str(identity),))
            c.execute("INSERT OR IGNORE INTO meta VALUES('cursor','0')")

    @contextlib.contextmanager
    def db(self):
        c=sqlite3.connect(self.state,timeout=35)
        c.row_factory=sqlite3.Row
        c.execute('PRAGMA foreign_keys=ON')
        c.execute('PRAGMA synchronous=FULL')
        try:
            with c: yield c
        finally:
            c.close()

    def request(self,path,data=None,key=None):
        headers={'Authorization':'Bearer '+self.token}
        if data is not None: headers['Content-Type']='application/json'
        if key: headers['Idempotency-Key']=key
        req=urllib.request.Request(BASE+'/discussions'+path,
            data=json.dumps(data).encode() if data is not None else None,headers=headers)
        with self.opener.open(req,timeout=25) as r:
            raw=r.read(1000001)
            if len(raw)>1000000: raise ValueError('Response too large')
            return json.loads(raw)

    def subscriptions(self,tid=None,enabled=None):
        if tid is None: return self.request('/api/subscriptions')
        return self.request('/api/subscriptions',{'thread_id':tid,'enabled':enabled,
            'consent':'subscribe' if enabled else None})

    def poll(self):
        # One process owns a pull at a time; crash rolls back the whole page.
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            cursor=int(c.execute("SELECT value FROM meta WHERE key='cursor'").fetchone()[0])
            page=self.request('/api/inbox?cursor='+str(cursor))
            end=page['next_cursor']; ids=[x['event_id'] for x in page['items']]
            if type(end) is not int or end<cursor or ids!=sorted(set(ids)) or any(type(i) is not int or not cursor<i<=end for i in ids):
                raise ValueError('Invalid event page')
            for event in page['items']:
                c.execute('INSERT OR IGNORE INTO events(id,payload) VALUES(?,?)',
                    (event['event_id'],json.dumps(event,ensure_ascii=False)))
                if event['state']!='public':
                    # Scrub earlier cached public text for the withdrawn message.
                    for prior in c.execute('SELECT id,payload FROM events').fetchall():
                        item=json.loads(prior['payload'])
                        if item['message_id']==event['message_id']:
                            item.pop('body',None);item.pop('alias',None);item['state']=event['state'];item['url']=None
                            c.execute('UPDATE events SET payload=? WHERE id=?',(json.dumps(item),prior['id']))
                            c.execute("UPDATE outbox SET state='held' WHERE event_id=? AND state='pending'",(prior['id'],))
            c.execute("UPDATE meta SET value=? WHERE key='cursor'",(str(end),))
            c.execute('INSERT OR IGNORE INTO pages VALUES(?)',(end,))
        return {'received':len(ids),'has_more':page['has_more'],'next_cursor':end,
                'poll_after_seconds':page.get('poll_after_seconds',300),'note':'Saved locally, not acknowledged or answered'}

    def pending(self):
        with self.db() as c:
            return [json.loads(r['payload']) for r in c.execute('SELECT payload FROM events WHERE done=0 ORDER BY id')]

    def done(self,event_id,reason):
        if not reason.strip() or len(reason)>1000: raise ValueError('A short decision is required')
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            job=c.execute('SELECT state FROM outbox WHERE event_id=?',(event_id,)).fetchone()
            if job and job['state'] not in ('sent','cancelled'):
                raise ValueError('Outgoing reply unresolved; send or explicitly cancel it first')
            if not c.execute('SELECT 1 FROM events WHERE id=?',(event_id,)).fetchone(): raise ValueError('Unknown event')
            c.execute('UPDATE events SET done=1,decision=? WHERE id=?',(reason,event_id))
        return {'processed':event_id}

    def reply(self,event_id,body):
        if not body.strip() or len(body)>4000: raise ValueError('Reply must contain 1–4000 characters')
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            row=c.execute('SELECT * FROM events WHERE id=?',(event_id,)).fetchone()
            if not row: raise ValueError('Unknown event')
            event=json.loads(row['payload'])
            if event['state']!='public' or event['kind']=='submission_status': raise ValueError('No public external message to reply to')
            payload=json.dumps({'body':body,'parent_id':event['message_id'],'consent':'publish'},ensure_ascii=False)
            old=c.execute('SELECT * FROM outbox WHERE event_id=?',(event_id,)).fetchone()
            if old:
                if old['payload']!=payload: raise ValueError('Reply already reserved with different text; do not change uncertain payload')
                return {'state':old['state'],'request_key':old['request_key']}
            if row['done']: raise ValueError('Event already processed')
            # One reply per source message, even if a later publication event repeats it.
            for job in c.execute('SELECT payload FROM outbox WHERE thread_id=?',(event['thread_id'],)):
                if json.loads(job['payload'])['parent_id']==event['message_id']:
                    raise ValueError('A reply to this message is already reserved')
            key=str(uuid.uuid4())
            c.execute('INSERT INTO outbox(event_id,request_key,thread_id,payload) VALUES(?,?,?,?)',
                (event_id,key,event['thread_id'],payload))
        return {'state':'pending','request_key':key,'note':'Saved, not sent. Use send explicitly.'}

    def send(self):
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            job=c.execute("SELECT * FROM outbox WHERE state='pending' ORDER BY event_id LIMIT 1").fetchone()
            if not job: return {'sent':False}
            # The key/payload were committed by reply(). Losing this receipt is safe to retry.
            receipt=self.request('/api/threads/'+str(job['thread_id'])+'/messages',json.loads(job['payload']),job['request_key'])
            if type(receipt.get('message_id')) is not int or receipt.get('thread_id')!=job['thread_id']:
                raise ValueError('Unexpected receipt; preserve pending request')
            c.execute("UPDATE outbox SET state='sent',receipt=? WHERE event_id=?",(json.dumps(receipt),job['event_id']))
            c.execute("UPDATE events SET done=1,decision='reply receipt saved' WHERE id=?",(job['event_id'],))
        return {'sent':True,'receipt':receipt}

    def cancel(self,event_id):
        with self.db() as c:
            row=c.execute('SELECT state FROM outbox WHERE event_id=?',(event_id,)).fetchone()
            if not row or row['state']=='sent': raise ValueError('No cancellable outgoing record')
            c.execute("UPDATE outbox SET state='cancelled' WHERE event_id=?",(event_id,))
        return {'cancelled':event_id,'notice':'Stops local retries only; uncertain remote publication may already exist. Key retained.'}

    def ack(self):
        with self.db() as c:
            c.execute('BEGIN IMMEDIATE')
            boundary=c.execute('''SELECT max(cursor) FROM pages WHERE NOT EXISTS
                (SELECT 1 FROM events WHERE done=0 AND id<=pages.cursor)''').fetchone()[0]
            if boundary is None: return {'acknowledged':False,'reason':'Unprocessed events remain'}
            result=self.request('/api/inbox/ack',{'cursor':boundary})
            c.execute('DELETE FROM pages WHERE cursor<=?',(boundary,))
        return result


def main():
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--state',default='open-archive-state.db',help='Keep this local SQLite file across restarts; do not share it')
    sub=parser.add_subparsers(dest='command',required=True)
    for command in ('poll','pending','send','ack','subscriptions'): sub.add_parser(command)
    for command in ('subscribe','unsubscribe'):
        p=sub.add_parser(command);p.add_argument('thread_id',type=int)
        if command=='subscribe': p.add_argument('--consent',action='store_true',required=True)
    p=sub.add_parser('done');p.add_argument('event_id',type=int);p.add_argument('--reason',required=True)
    p=sub.add_parser('reply');p.add_argument('event_id',type=int);p.add_argument('--body-file',required=True);p.add_argument('--consent',action='store_true',required=True)
    p=sub.add_parser('cancel');p.add_argument('event_id',type=int)
    args=parser.parse_args()
    try:
        client=Client(args.state,os.environ.get('OPEN_ARCHIVE_TOKEN'))
        if args.command in ('subscribe','unsubscribe'): result=client.subscriptions(args.thread_id,args.command=='subscribe')
        elif args.command=='reply': result=client.reply(args.event_id,pathlib.Path(args.body_file).read_text(encoding='utf-8'))
        elif args.command=='done': result=client.done(args.event_id,args.reason)
        elif args.command=='cancel': result=client.cancel(args.event_id)
        else: result=getattr(client,args.command)()
        print(json.dumps(result,ensure_ascii=False,indent=2))
    except Exception as exc:
        # Never print request headers, token or untrusted server error bodies.
        code=exc.code if isinstance(exc,urllib.error.HTTPError) else type(exc).__name__
        parser.exit(1,'Request not completed ('+str(code)+'). Keep state and the original outgoing key; inspect before retrying.\n')


if __name__=='__main__': main()