"""Synchronous owner-scoped SQLite projections; routes must offload calls.

Missing/foreign objects raise KeyError; invalid input ValueError; conflicting
revision, idempotency payload or terminal transition ChatConflict. Branch with
edited user content copies BEFORE that user, without submitting replacement:
caller must send the edited turn. Otherwise copy through the selected message.
Only completed role/content history seeds a fresh Hermes session; no tools rerun.
"""
from __future__ import annotations

import json
import time
import uuid


class ChatConflict(ValueError):
    """Optimistic revision, idempotency payload or state conflict (HTTP 409)."""


def text(value, name, maximum, empty=False):
    if not isinstance(value, str) or len(value) > maximum or (not empty and not value.strip()) or '\x00' in value:
        raise ValueError(f'Invalid {name}')
    try:
        if len(value.encode('utf-8')) > maximum * 4:
            raise ValueError(f'Invalid {name}')
    except UnicodeError as exc:
        raise ValueError(f'Invalid {name}') from exc
    return value


def integer(value, name, minimum=0, maximum=2**63-1):
    if type(value) is not int or not minimum <= value <= maximum:
        raise ValueError(f'Invalid {name}')
    return value


class ChatStore:
    TERMINAL = {'completed', 'failed', 'cancelled'}

    def __init__(self, accounts):
        self.accounts, self.db, self.lock = accounts, accounts.db, accounts.lock
        with self.lock, self.db:
            self.db.execute('PRAGMA foreign_keys=ON')
            self.db.executescript('''
            CREATE TABLE IF NOT EXISTS chat_workspaces (
              id TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
              name TEXT NOT NULL, instructions TEXT NOT NULL, revision INTEGER NOT NULL,
              created_at REAL NOT NULL, updated_at REAL NOT NULL, UNIQUE(id,owner));
            CREATE TABLE IF NOT EXISTS chat_conversations (
              id TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
              title TEXT NOT NULL, workspace_id TEXT, pinned INTEGER NOT NULL DEFAULT 0,
              pin_rank INTEGER NOT NULL DEFAULT 0, archived INTEGER NOT NULL DEFAULT 0,
              temporary INTEGER NOT NULL DEFAULT 0, revision INTEGER NOT NULL,
              created_at REAL NOT NULL, updated_at REAL NOT NULL,
              hermes_session_id TEXT, hermes_seeded INTEGER NOT NULL DEFAULT 0,
              UNIQUE(id,owner), FOREIGN KEY(workspace_id,owner) REFERENCES chat_workspaces(id,owner));
            CREATE TABLE IF NOT EXISTS chat_messages (
              id TEXT PRIMARY KEY, owner TEXT NOT NULL, conversation_id TEXT NOT NULL,
              role TEXT NOT NULL CHECK(role IN ('user','assistant')), content TEXT NOT NULL,
              status TEXT NOT NULL, run_id TEXT, created_at REAL NOT NULL,
              position INTEGER NOT NULL, UNIQUE(conversation_id,position),
              FOREIGN KEY(conversation_id,owner) REFERENCES chat_conversations(id,owner) ON DELETE CASCADE,
              FOREIGN KEY(run_id,owner) REFERENCES chat_runs(id,owner) ON DELETE CASCADE);
            CREATE TABLE IF NOT EXISTS chat_runs (
              id TEXT PRIMARY KEY, owner TEXT NOT NULL, conversation_id TEXT NOT NULL,
              status TEXT NOT NULL, message_id TEXT NOT NULL, created_at REAL NOT NULL,
              updated_at REAL NOT NULL, error TEXT, UNIQUE(id,owner),
              FOREIGN KEY(conversation_id,owner) REFERENCES chat_conversations(id,owner) ON DELETE CASCADE,
              FOREIGN KEY(message_id) REFERENCES chat_messages(id) DEFERRABLE INITIALLY DEFERRED);
            CREATE TABLE IF NOT EXISTS chat_idempotency (
              owner TEXT NOT NULL, conversation_id TEXT NOT NULL, key TEXT NOT NULL,
              payload TEXT NOT NULL, message_id TEXT NOT NULL REFERENCES chat_messages(id) ON DELETE CASCADE,
              run_id TEXT NOT NULL REFERENCES chat_runs(id) ON DELETE CASCADE,
              PRIMARY KEY(owner,conversation_id,key),
              FOREIGN KEY(conversation_id,owner) REFERENCES chat_conversations(id,owner) ON DELETE CASCADE);
            CREATE TABLE IF NOT EXISTS chat_events (
              owner TEXT NOT NULL, run_id TEXT NOT NULL, sequence INTEGER NOT NULL,
              event TEXT NOT NULL, data TEXT NOT NULL, PRIMARY KEY(run_id,sequence),
              FOREIGN KEY(run_id,owner) REFERENCES chat_runs(id,owner) ON DELETE CASCADE);
            CREATE TABLE IF NOT EXISTS chat_preferences (
              owner TEXT PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
              palette TEXT NOT NULL, appearance TEXT NOT NULL, revision INTEGER NOT NULL);
            CREATE TABLE IF NOT EXISTS chat_changes (
              sequence INTEGER PRIMARY KEY AUTOINCREMENT,
              owner TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
              kind TEXT NOT NULL, object_id TEXT NOT NULL, deleted INTEGER NOT NULL,
              data TEXT, created_at REAL NOT NULL);
            CREATE INDEX IF NOT EXISTS chat_changes_owner ON chat_changes(owner,sequence);
            CREATE INDEX IF NOT EXISTS chat_conversation_owner ON chat_conversations(owner,updated_at,id);
            CREATE INDEX IF NOT EXISTS chat_message_order ON chat_messages(conversation_id,position);
            ''')
            columns = {r[1] for r in self.db.execute('PRAGMA table_info(chat_conversations)')}
            for column, definition in [('hermes_session_id', 'TEXT'), ('hermes_seeded', 'INTEGER NOT NULL DEFAULT 0')]:
                if column not in columns:
                    self.db.execute(f'ALTER TABLE chat_conversations ADD COLUMN {column} {definition}')
            if 'attachments' not in {r[1] for r in self.db.execute('PRAGMA table_info(chat_messages)')}:
                self.db.execute("ALTER TABLE chat_messages ADD COLUMN attachments TEXT NOT NULL DEFAULT '[]'")

    def _owner(self, owner):
        text(owner, 'owner', 128)
        if not self.db.execute("SELECT 1 FROM users WHERE id=? AND desired!='deleted'", (owner,)).fetchone():
            raise KeyError(owner)

    def _quota(self, table, owner, maximum):
        if self.db.execute(f'SELECT count(*) FROM {table} WHERE owner=?',(owner,)).fetchone()[0]>=maximum:
            raise ValueError('Account chat storage limit reached')

    def _row(self, table, owner, identity):
        self._owner(owner)
        text(identity, 'id', 128)
        row = self.db.execute(f'SELECT * FROM {table} WHERE owner=? AND id=?', (owner, identity)).fetchone()
        if row is None:
            raise KeyError(identity)
        return dict(row)

    @staticmethod
    def _conversation(row):
        result = {k: row[k] for k in ('id','title','workspace_id','pinned','pin_rank','archived','temporary','revision','created_at','updated_at')}
        for k in ('pinned', 'archived', 'temporary'):
            result[k] = bool(result[k])
        return result

    @staticmethod
    def _message(row):
        return {**{k: row[k] for k in ('id','conversation_id','role','content','status','run_id','created_at')}, 'attachments': json.loads(row['attachments']) if 'attachments' in row.keys() else []}

    @staticmethod
    def _run(row):
        return {k: row[k] for k in ('id','conversation_id','status','message_id','created_at','updated_at','error')}

    @staticmethod
    def _workspace(row):
        return {k: row[k] for k in ('id','name','instructions','revision','created_at','updated_at')}

    def _touch(self, owner, cid):
        self.db.execute('UPDATE chat_conversations SET revision=revision+1,updated_at=? WHERE id=? AND owner=?',(time.time(),cid,owner))
        self._change(owner,'conversation',cid,self._conversation(self._row('chat_conversations',owner,cid)))

    def _change(self, owner, kind, identity, data=None):
        self._quota('chat_changes',owner,100000)
        self.db.execute('INSERT INTO chat_changes(owner,kind,object_id,deleted,data,created_at) VALUES(?,?,?,?,?,?)',
                        (owner, kind, identity, data is None, json.dumps(data) if data is not None else None, time.time()))

    def create_conversation(self, owner, title='New chat', workspace_id=None, temporary=False):
        text(title, 'title', 200)
        if type(temporary) is not bool or temporary:
            raise ValueError('Temporary conversations are not enabled')
        with self.lock, self.db:
            self._owner(owner)
            if workspace_id is not None:
                self._row('chat_workspaces', owner, workspace_id)
            self._quota('chat_conversations',owner,1000)
            identity, now = uuid.uuid4().hex, time.time()
            self.db.execute('INSERT INTO chat_conversations(id,owner,title,workspace_id,temporary,revision,created_at,updated_at) VALUES(?,?,?,?,?,1,?,?)',
                            (identity, owner, title, workspace_id, temporary, now, now))
            result = self._conversation(self._row('chat_conversations', owner, identity))
            self._change(owner, 'conversation', identity, result)
            return result

    @staticmethod
    def _cursor(cursor):
        if cursor is None:
            return 0
        if not isinstance(cursor, str) or not cursor.isascii() or not cursor.isdigit() or len(cursor)>9:
            raise ValueError('Invalid cursor')
        return int(cursor)

    def list_conversations(self, owner, q='', workspace_id=None, archived=False, cursor=None, limit=50):
        text(q, 'search', 200, empty=True)
        integer(limit, 'limit', 1, 200)
        if type(archived) is not bool:
            raise ValueError('Invalid archived')
        offset = self._cursor(cursor)
        with self.lock:
            self._owner(owner)
            clauses, params = ['owner=?', 'archived=?'], [owner, archived]
            if workspace_id is not None:
                self._row('chat_workspaces', owner, workspace_id)
                clauses.append('workspace_id=?')
                params.append(workspace_id)
            if q:
                clauses.append('(instr(lower(title),lower(?))>0 OR EXISTS (SELECT 1 FROM chat_messages m WHERE m.conversation_id=chat_conversations.id AND instr(lower(m.content),lower(?))>0))')
                params.extend([q, q])
            rows = list(self.db.execute('SELECT * FROM chat_conversations WHERE ' + ' AND '.join(clauses) + ' ORDER BY pinned DESC,pin_rank,updated_at DESC,id LIMIT ? OFFSET ?', (*params, limit+1, offset)))
            return {'conversations': [self._conversation(r) for r in rows[:limit]], 'next_cursor': str(offset+limit) if len(rows)>limit else None}

    def get_conversation(self, owner, cid, cursor=None, limit=100):
        integer(limit, 'limit', 1, 500)
        offset = self._cursor(cursor)
        with self.lock:
            conversation = self._conversation(self._row('chat_conversations', owner, cid))
            if cursor is None:
                rows = list(self.db.execute('SELECT * FROM chat_messages WHERE owner=? AND conversation_id=? ORDER BY position DESC LIMIT ?', (owner,cid,limit+1)))
            else:
                rows = list(self.db.execute('SELECT * FROM chat_messages WHERE owner=? AND conversation_id=? AND position<? ORDER BY position DESC LIMIT ?', (owner,cid,offset,limit+1)))
            runs = [self._run(r) for r in self.db.execute('SELECT * FROM chat_runs WHERE owner=? AND conversation_id=? ORDER BY created_at,id', (owner,cid))]
            return {'conversation': conversation, 'messages': [self._message(r) for r in reversed(rows[:limit])], 'runs': runs, 'next_cursor': str(rows[limit-1]['position']) if len(rows)>limit else None}

    def update_conversation(self, owner, cid, revision, **fields):
        integer(revision, 'revision', 1)
        if not fields or not fields.keys() <= {'title','workspace_id','pinned','pin_rank','archived'}:
            raise ValueError('Invalid conversation update')
        for key,value in fields.items():
            if key=='title': text(value, key, 200)
            if key in ('pinned','archived') and type(value) is not bool: raise ValueError(f'Invalid {key}')
            if key=='pin_rank': integer(value,key,0,1000000)
        with self.lock, self.db:
            row = self._row('chat_conversations',owner,cid)
            if row['revision']!=revision: raise ChatConflict('Stale revision')
            if fields.get('workspace_id') is not None: self._row('chat_workspaces',owner,fields['workspace_id'])
            self.db.execute('UPDATE chat_conversations SET '+','.join(f'{k}=?' for k in fields)+',revision=revision+1,updated_at=? WHERE id=? AND owner=?', (*fields.values(),time.time(),cid,owner))
            result = self._conversation(self._row('chat_conversations',owner,cid))
            self._change(owner,'conversation',cid,result)
            return result

    def delete_conversation(self, owner, cid):
        with self.lock, self.db:
            self._row('chat_conversations',owner,cid)
            if self.db.execute("SELECT 1 FROM chat_runs WHERE owner=? AND conversation_id=? AND status IN ('queued','running','awaiting_approval') LIMIT 1",(owner,cid)).fetchone():
                raise ChatConflict('Execution still active')
            if self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='chat_transport'").fetchone():
                for row in self.db.execute('SELECT metadata FROM chat_transport WHERE owner=? AND conversation_id=?',(owner,cid)):
                    metadata=json.loads(row['metadata'])
                    if metadata.get('upstream_id') and not metadata.get('remote_terminal'):
                        raise ChatConflict('Remote execution still active')
                self.db.execute('DELETE FROM chat_transport WHERE owner=? AND conversation_id=?',(owner,cid))
            self.db.execute('DELETE FROM chat_conversations WHERE id=? AND owner=?',(cid,owner))
            self._change(owner,'conversation',cid)
            return {'ok': True}

    def create_workspace(self, owner, name, instructions=''):
        text(name,'name',100); text(instructions,'instructions',16000,empty=True)
        with self.lock, self.db:
            self._owner(owner)
            self._quota('chat_workspaces',owner,100)
            identity,now = uuid.uuid4().hex,time.time()
            self.db.execute('INSERT INTO chat_workspaces VALUES(?,?,?,?,1,?,?)',(identity,owner,name,instructions,now,now))
            result=self._workspace(self._row('chat_workspaces',owner,identity))
            self._change(owner,'workspace',identity,result)
            return result

    def list_workspaces(self, owner):
        with self.lock:
            self._owner(owner)
            return [self._workspace(r) for r in self.db.execute('SELECT * FROM chat_workspaces WHERE owner=? ORDER BY created_at,id',(owner,))]

    def get_workspace(self, owner, wid):
        with self.lock: return self._workspace(self._row('chat_workspaces',owner,wid))

    def update_workspace(self, owner, wid, revision, **fields):
        integer(revision,'revision',1)
        if not fields or not fields.keys()<={'name','instructions'}: raise ValueError('Invalid workspace update')
        for key,value in fields.items(): text(value,key,100 if key=='name' else 16000,empty=key=='instructions')
        with self.lock,self.db:
            row=self._row('chat_workspaces',owner,wid)
            if row['revision']!=revision: raise ChatConflict('Stale revision')
            self.db.execute('UPDATE chat_workspaces SET '+','.join(f'{k}=?' for k in fields)+',revision=revision+1,updated_at=? WHERE id=? AND owner=?',(*fields.values(),time.time(),wid,owner))
            result=self._workspace(self._row('chat_workspaces',owner,wid)); self._change(owner,'workspace',wid,result)
            return result

    def delete_workspace(self, owner, wid):
        with self.lock,self.db:
            self._row('chat_workspaces',owner,wid)
            ids=[r[0] for r in self.db.execute('SELECT id FROM chat_conversations WHERE owner=? AND workspace_id=?',(owner,wid))]
            self.db.execute('UPDATE chat_conversations SET workspace_id=NULL,revision=revision+1,updated_at=? WHERE owner=? AND workspace_id=?',(time.time(),owner,wid))
            for cid in ids: self._change(owner,'conversation',cid,self._conversation(self._row('chat_conversations',owner,cid)))
            self.db.execute('DELETE FROM chat_workspaces WHERE owner=? AND id=?',(owner,wid)); self._change(owner,'workspace',wid)
            return {'ok':True}

    def get_preferences(self, owner):
        with self.lock:
            self._owner(owner)
            row=self.db.execute('SELECT palette,appearance,revision FROM chat_preferences WHERE owner=?',(owner,)).fetchone()
            return dict(row) if row else {'palette':'opal','appearance':'system','revision':0}

    def set_preferences(self, owner, revision, palette, appearance):
        integer(revision,'revision')
        if not isinstance(palette,str) or palette not in ('opal','graphite','ocean','sage','lavender'): raise ValueError('Invalid palette')
        if not isinstance(appearance,str) or appearance not in ('system','light','dark'): raise ValueError('Invalid appearance')
        with self.lock,self.db:
            if self.get_preferences(owner)['revision']!=revision: raise ChatConflict('Stale revision')
            self.db.execute('INSERT INTO chat_preferences VALUES(?,?,?,?) ON CONFLICT(owner) DO UPDATE SET palette=excluded.palette,appearance=excluded.appearance,revision=excluded.revision',(owner,palette,appearance,revision+1))
            result=self.get_preferences(owner); self._change(owner,'preferences',owner,result)
            return result

    def sync(self, owner, after=0, limit=200):
        integer(after,'after'); integer(limit,'limit',1,500)
        with self.lock:
            self._owner(owner)
            rows=list(self.db.execute('SELECT * FROM chat_changes WHERE owner=? AND sequence>? ORDER BY sequence LIMIT ?',(owner,after,limit)))
            changes=[{'sequence':r['sequence'],'kind':r['kind'],'id':r['object_id'],'deleted':bool(r['deleted']),'data':json.loads(r['data']) if r['data'] is not None else None,'created_at':r['created_at']} for r in rows]
            return {'changes':changes,'cursor':rows[-1]['sequence'] if rows else after}

    def _insert_message(self,owner,cid,role,content,status,run_id=None,identity=None):
        identity=identity or uuid.uuid4().hex
        position=self.db.execute('SELECT COALESCE(MAX(position),0)+1 FROM chat_messages WHERE conversation_id=?',(cid,)).fetchone()[0]
        self._quota('chat_messages', owner, 20000)
        self.db.execute('INSERT INTO chat_messages(id,owner,conversation_id,role,content,status,run_id,created_at,position) VALUES(?,?,?,?,?,?,?,?,?)',(identity,owner,cid,role,content,status,run_id,time.time(),position))
        return self._message(self._row('chat_messages',owner,identity))

    def send(self, owner, cid, content, key, attachment_ids=None, mode='normal'):
        """Atomic admission; replay returns same IDs and never creates new work."""
        text(content,'content',32000); text(key,'idempotency key',200)
        if attachment_ids is None: attachment_ids=[]
        if not isinstance(attachment_ids,list) or len(attachment_ids)>8: raise ValueError('Invalid attachments')
        for attachment_id in attachment_ids: text(attachment_id,'attachment id',128)
        if len(set(attachment_ids))!=len(attachment_ids): raise ValueError('Duplicate attachments')
        if not isinstance(mode,str) or mode not in ('normal','research'): raise ValueError('Invalid mode')
        payload=json.dumps({'content':content,'attachment_ids':attachment_ids,'mode':mode},sort_keys=True)
        with self.lock,self.db:
            self._row('chat_conversations',owner,cid)
            prior=self.db.execute('SELECT * FROM chat_idempotency WHERE owner=? AND conversation_id=? AND key=?',(owner,cid,key)).fetchone()
            if prior:
                if prior['payload']!=payload: raise ChatConflict('Idempotency payload changed')
                return {'message':self._message(self._row('chat_messages',owner,prior['message_id'])),'run':self._run(self._row('chat_runs',owner,prior['run_id']))}
            for attachment_id in attachment_ids:
                exists=self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='hope_attachments'").fetchone()
                if not exists or not self.db.execute('SELECT 1 FROM hope_attachments WHERE owner=? AND id=? AND deleted_at IS NULL',(owner,attachment_id)).fetchone(): raise KeyError(attachment_id)
            self._quota('chat_runs',owner,10000)
            mid,rid,now=uuid.uuid4().hex,uuid.uuid4().hex,time.time()
            self.db.execute('INSERT INTO chat_runs VALUES(?,?,?,?,?,?,?,?)',(rid,owner,cid,'queued',mid,now,now,None))
            self._insert_message(owner,cid,'user',content,'completed',rid,mid)
            self.db.execute('UPDATE chat_messages SET attachments=? WHERE id=?',(json.dumps(attachment_ids),mid))
            message=self._message(self._row('chat_messages',owner,mid))
            self.db.execute('INSERT INTO chat_idempotency VALUES(?,?,?,?,?,?)',(owner,cid,key,payload,mid,rid))
            self.db.execute('UPDATE chat_conversations SET revision=revision+1,updated_at=? WHERE id=?',(now,cid))
            self._change(owner,'conversation',cid,self._conversation(self._row('chat_conversations',owner,cid)))
            return {'message':message,'run':self._run(self._row('chat_runs',owner,rid))}

    def get_run(self, owner, rid):
        with self.lock:
            run=self._run(self._row('chat_runs',owner,rid))
            return {'run':run,'messages':[self._message(r) for r in self.db.execute('SELECT * FROM chat_messages WHERE owner=? AND run_id=? ORDER BY position',(owner,rid))]}

    def update_run(self, owner, rid, status, error=None):
        transitions={'queued':{'running','failed','cancelled'},'running':{'awaiting_approval','completed','failed','cancelled'},'awaiting_approval':{'running','failed','cancelled'}}
        if not isinstance(status,str) or status not in {*transitions,*self.TERMINAL}: raise ValueError('Invalid run status')
        if error is not None: text(error,'error',1000,empty=True)
        with self.lock,self.db:
            row=self._row('chat_runs',owner,rid)
            if row['status']==status:
                if row['error']!=error: raise ChatConflict('Run already has this status')
                return self._run(row)
            if status not in transitions.get(row['status'],set()): raise ChatConflict('Invalid run transition')
            self.db.execute('UPDATE chat_runs SET status=?,error=?,updated_at=? WHERE id=? AND owner=?',(status,error,time.time(),rid,owner))
            if status in self.TERMINAL: self._touch(owner,row['conversation_id'])
            return self._run(self._row('chat_runs',owner,rid))

    def add_message(self, owner, cid, role, content, status='completed', run_id=None):
        text(content,'content',32000,empty=True)
        if not isinstance(role,str) or role not in ('user','assistant') or not isinstance(status,str) or status not in ('pending','streaming','completed','failed','cancelled'): raise ValueError('Invalid message')
        with self.lock,self.db:
            self._row('chat_conversations',owner,cid)
            if run_id is not None:
                run=self._row('chat_runs',owner,run_id)
                if run['conversation_id']!=cid: raise KeyError(run_id)
                if run['status'] in self.TERMINAL: raise ChatConflict('Run finished')
            result=self._insert_message(owner,cid,role,content,status,run_id)
            if status in self.TERMINAL: self._touch(owner,cid)
            return result

    def update_message(self, owner, mid, content=None, status=None):
        if content is not None: text(content,'content',32000,empty=True)
        if status is not None and (not isinstance(status,str) or status not in ('pending','streaming','completed','failed','cancelled')): raise ValueError('Invalid message status')
        with self.lock,self.db:
            row=self._row('chat_messages',owner,mid)
            new_content=row['content'] if content is None else content
            new_status=row['status'] if status is None else status
            transitions={'pending':{'streaming','completed','failed','cancelled'},'streaming':self.TERMINAL}
            if row['status'] in self.TERMINAL and (new_content!=row['content'] or new_status!=row['status']): raise ChatConflict('Message finished')
            if new_status!=row['status'] and new_status not in transitions.get(row['status'],set()): raise ChatConflict('Invalid message transition')
            if row['run_id'] and new_status in ('pending','streaming','completed'):
                run=self._row('chat_runs',owner,row['run_id'])
                if run['status'] in ('failed','cancelled') and (new_content!=row['content'] or new_status!=row['status']): raise ChatConflict('Run finished')
            self.db.execute('UPDATE chat_messages SET content=?,status=? WHERE owner=? AND id=?',(new_content,new_status,owner,mid))
            if new_status in self.TERMINAL and (new_content!=row['content'] or new_status!=row['status']): self._touch(owner,row['conversation_id'])
            return self._message(self._row('chat_messages',owner,mid))

    def append_event(self, owner, rid, event, data):
        if not isinstance(event,str) or event not in ('message','run','progress','approval','done'): raise ValueError('Invalid event')
        if not isinstance(data,dict): raise ValueError('Invalid event data')
        try: encoded=json.dumps(data,allow_nan=False)
        except (ValueError,TypeError) as exc: raise ValueError('Invalid event data') from exc
        if len(encoded.encode())>131072: raise ValueError('Event too large')
        with self.lock,self.db:
            self._row('chat_runs',owner,rid)
            budget=self.db.execute('SELECT count(*),COALESCE(SUM(length(CAST(data AS BLOB))),0) FROM chat_events WHERE run_id=?',(rid,)).fetchone()
            # Reserve terminal event slots so capacity failures can terminate cleanly.
            ceiling=10010 if event in ('run','done') else 10000
            if budget[0]>=ceiling or budget[1]+len(encoded.encode())>16*1024*1024+(131072 if event in ('run','done') else 0):
                raise ValueError('Run event storage limit reached')
            sequence=self.db.execute('SELECT COALESCE(MAX(sequence),0)+1 FROM chat_events WHERE run_id=?',(rid,)).fetchone()[0]
            self.db.execute('INSERT INTO chat_events VALUES(?,?,?,?,?)',(owner,rid,sequence,event,encoded))
            return {'sequence':sequence,'event':event,'data':json.loads(encoded)}

    def events(self, owner, rid, after=0, limit=500):
        integer(after,'after'); integer(limit,'limit',1,1000)
        with self.lock:
            self._row('chat_runs',owner,rid)
            return [{'sequence':r['sequence'],'event':r['event'],'data':json.loads(r['data'])} for r in self.db.execute('SELECT * FROM chat_events WHERE owner=? AND run_id=? AND sequence>? ORDER BY sequence LIMIT ?',(owner,rid,after,limit))]

    def branch(self, owner, cid, message_id, content=None, title=None):
        if content is not None: text(content,'content',32000)
        if title is not None: text(title,'title',200)
        with self.lock,self.db:
            parent=self._row('chat_conversations',owner,cid)
            cutoff=self._row('chat_messages',owner,message_id)
            if cutoff['conversation_id']!=cid: raise KeyError(message_id)
            if content is not None and cutoff['role']!='user': raise ValueError('Only user messages can be edited')
            if cutoff['status']!='completed': raise ChatConflict('Cannot branch unfinished message')
            self._quota('chat_conversations',owner,1000)
            identity,now=uuid.uuid4().hex,time.time()
            self.db.execute('INSERT INTO chat_conversations(id,owner,title,workspace_id,temporary,revision,created_at,updated_at) VALUES(?,?,?,?,?,1,?,?)',(identity,owner,title or parent['title'],parent['workspace_id'],parent['temporary'],now,now))
            comparator='<' if content is not None else '<='
            rows=list(self.db.execute(f'SELECT * FROM chat_messages WHERE owner=? AND conversation_id=? AND position{comparator}? ORDER BY position',(owner,cid,cutoff['position'])))
            for row in rows:
                if row['status']=='completed':
                    copied=self._insert_message(owner,identity,row['role'],row['content'],'completed')
                    self.db.execute('UPDATE chat_messages SET attachments=? WHERE id=?',(row['attachments'],copied['id']))
            result=self._conversation(self._row('chat_conversations',owner,identity)); self._change(owner,'conversation',identity,result)
            return result

    def history(self, owner, cid):
        """Completed branch-local turns for one-time Hermes history seeding."""
        with self.lock:
            self._row('chat_conversations',owner,cid)
            return [{'role':r['role'],'content':r['content']} for r in self.db.execute("SELECT role,content FROM chat_messages WHERE owner=? AND conversation_id=? AND status='completed' ORDER BY position",(owner,cid))]

    def get_session(self, owner, cid):
        """Trusted executor only; never include this result in public JSON."""
        with self.lock:
            row=self._row('chat_conversations',owner,cid)
            return {'hermes_session_id':row['hermes_session_id'],'hermes_seeded':bool(row['hermes_seeded'])}

    def set_session(self, owner, cid, hermes_session_id, seeded=False):
        text(hermes_session_id,'Hermes session',200)
        if type(seeded) is not bool: raise ValueError('Invalid seeded flag')
        with self.lock,self.db:
            row=self._row('chat_conversations',owner,cid)
            if row['hermes_session_id'] is not None and row['hermes_session_id']!=hermes_session_id: raise ChatConflict('Session already assigned')
            if row['hermes_seeded'] and not seeded: raise ChatConflict('Session already seeded')
            self.db.execute('UPDATE chat_conversations SET hermes_session_id=?,hermes_seeded=? WHERE owner=? AND id=?',(hermes_session_id,seeded,owner,cid))
            return {'hermes_session_id':hermes_session_id,'hermes_seeded':seeded}

    def get_hermes_session(self, owner, cid):
        return self.get_session(owner,cid)['hermes_session_id']

    def set_hermes_session(self, owner, cid, session_id, seeded=True):
        return self.set_session(owner,cid,session_id,seeded)

    def active_runs(self, owner=None):
        """Trusted recovery listing; None lists all owners, with private owner ID."""
        with self.lock:
            if owner is not None: self._owner(owner)
            where="status IN ('queued','running','awaiting_approval')"
            params=()
            if owner is not None: where+=' AND owner=?'; params=(owner,)
            return [{**self._run(r),'owner':r['owner']} for r in self.db.execute('SELECT * FROM chat_runs WHERE '+where+' ORDER BY created_at,id',params)]

    def claim_run(self, owner, rid):
        """Only queued -> running admits execution; a retry cannot claim twice."""
        with self.lock,self.db:
            row=self._row('chat_runs',owner,rid)
            if row['status']!='queued': return False
            self.db.execute("UPDATE chat_runs SET status='running',updated_at=? WHERE id=? AND owner=?",(time.time(),rid,owner))
            return True

    def get_message(self, owner, mid):
        with self.lock: return self._message(self._row('chat_messages',owner,mid))

    get_events = events
    get_history = history
    conversation_history = history
    branch_conversation = branch
    update_preferences = set_preferences
    list_events = events
    pending_runs = active_runs