feat: add durable owner-scoped chat store

control/chat_store.py

@@ -0,0 +1,468 @@
+"""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)
+            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):
+        """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')
+        payload=json.dumps({'content':content,'attachment_ids':attachment_ids},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

tests/unit/test_chat_store.py

@@ -0,0 +1,161 @@
+"""Offline chat persistence tests; never call deployed services."""
+import sqlite3
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from control.accounts import Accounts
+from control.chat_store import ChatConflict, ChatStore
+
+
+@pytest.fixture
+def setup(tmp_path):
+    accounts = Accounts(str(tmp_path / 'accounts.db'))
+    a = accounts.create('alice', 'Alice', 'admin', 'correct-password-alice')['id']
+    b = accounts.create('bob', 'Bob', 'user', 'correct-password-bob')['id']
+    store = ChatStore(accounts)
+    yield accounts, store, a, b
+    accounts.close()
+
+
+def test_metadata_ownership_revisions_and_tombstones(setup):
+    accounts, s, a, b = setup
+    assert s.db is accounts.db and s.lock is accounts.lock
+    w = s.create_workspace(a, 'Work', 'Be concise')
+    c = s.create_conversation(a, workspace_id=w['id'])
+    assert 'hermes_session_id' not in c
+    with pytest.raises(KeyError): s.get_conversation(b, c['id'])
+    with pytest.raises(KeyError): s.create_conversation(b, workspace_id=w['id'])
+    with pytest.raises(KeyError): s.delete_workspace(b, w['id'])
+    c = s.update_conversation(a, c['id'], c['revision'], pinned=True, title='Revised')
+    with pytest.raises(ChatConflict): s.update_conversation(a, c['id'], 1, title='Stale')
+    w = s.update_workspace(a, w['id'], 1, name='Renamed')
+    with pytest.raises(ChatConflict): s.update_workspace(a, w['id'], 1, name='Stale')
+    s.delete_workspace(a, w['id'])
+    assert s.get_conversation(a, c['id'])['conversation']['workspace_id'] is None
+    s.delete_conversation(a, c['id'])
+    changes = s.sync(a)['changes']
+    assert any(x['kind']=='workspace' and x['deleted'] for x in changes)
+    assert changes[-1]['deleted'] and changes[-1]['id']==c['id']
+    assert s.sync(b)['changes']==[]
+    assert s.sync(a, s.sync(a)['cursor'])['changes']==[]
+
+
+def test_preferences_and_validation(setup):
+    _, s, a, b = setup
+    assert s.get_preferences(a)['revision']==0
+    assert s.set_preferences(a, 0, 'ocean', 'dark')['revision']==1
+    with pytest.raises(ChatConflict): s.set_preferences(a, 0, 'opal', 'light')
+    assert s.get_preferences(b)['palette']=='opal'
+    invalid = [lambda:s.create_conversation(a, title=''), lambda:s.create_conversation(a, temporary=1),
+               lambda:s.create_workspace(a, 'x', instructions=7), lambda:s.list_conversations(a, limit=True),
+               lambda:s.list_conversations(a, cursor='-1'), lambda:s.set_preferences(a, True, 'opal', 'system'),
+               lambda:s.set_preferences(a, 1, [], 'dark')]
+    for call in invalid:
+        with pytest.raises(ValueError): call()
+
+
+def test_send_durable_idempotency_concurrent_claim_and_fk(setup):
+    accounts, s, a, b = setup
+    cid = s.create_conversation(a)['id']
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        replies = list(pool.map(lambda _:s.send(a,cid,'hello','unique-key'), range(16)))
+    rid = replies[0]['run']['id']
+    assert len({x['run']['id'] for x in replies})==1
+    assert len(s.get_conversation(a,cid)['messages'])==1
+    with pytest.raises(ChatConflict): s.send(a,cid,'different','unique-key')
+    with pytest.raises(KeyError): s.send(b,cid,'hello','unique-key')
+    with pytest.raises(KeyError): s.send(a,cid,'hello','attachments',['fake'])
+    s2 = ChatStore(accounts)
+    assert s2.send(a,cid,'hello','unique-key')['run']['id']==rid
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        assert sum(pool.map(lambda _:s.claim_run(a,rid),range(8)))==1
+    assert accounts.db.execute('PRAGMA foreign_keys').fetchone()[0]==1
+    other = s.create_conversation(b)['id']
+    with pytest.raises(sqlite3.IntegrityError), accounts.db:
+        accounts.db.execute('UPDATE chat_conversations SET workspace_id=? WHERE id=?',('missing',other))
+    s.delete_conversation(a,cid)
+    assert accounts.db.execute('SELECT count(*) FROM chat_runs WHERE id=?',(rid,)).fetchone()[0]==0
+    assert accounts.db.execute('PRAGMA foreign_key_check').fetchall()==[]
+
+
+def test_run_messages_events_and_terminal_states(setup):
+    _, s, a, b = setup
+    cid = s.create_conversation(a)['id']
+    rid = s.send(a,cid,'hello','one')['run']['id']
+    s.update_run(a,rid,'running')
+    m = s.add_message(a,cid,'assistant','','pending',rid)
+    s.update_message(a,m['id'],content='part',status='streaming')
+    e1 = s.append_event(a,rid,'message',{'content':'part'})
+    e2 = s.append_event(a,rid,'progress',{'label':'Working'})
+    assert [e1['sequence'],e2['sequence']]==[1,2]
+    assert s.events(a,rid,after=1)==[e2]
+    with pytest.raises(KeyError): s.events(b,rid)
+    with pytest.raises(KeyError): s.update_message(b,m['id'],content='steal')
+    s.update_run(a,rid,'cancelled')
+    with pytest.raises(ChatConflict): s.update_run(a,rid,'running')
+    with pytest.raises(ChatConflict): s.update_run(a,rid,'completed')
+    with pytest.raises(ChatConflict): s.update_message(a,m['id'],content='late',status='completed')
+    s.update_message(a,m['id'],status='cancelled')
+    with pytest.raises(ChatConflict): s.update_message(a,m['id'],content='late')
+    assert s.get_run(a,rid)['run']['status']=='cancelled'
+
+
+def test_branch_copies_history_without_execution_or_session(setup):
+    _, s, a, b = setup
+    cid = s.create_conversation(a)['id']
+    u = s.add_message(a,cid,'user','original')
+    reply = s.add_message(a,cid,'assistant','answer')
+    s.add_message(a,cid,'user','later')
+    s.set_hermes_session(a,cid,'trusted-session')
+    branch = s.branch(a,cid,reply['id'])
+    assert s.history(a,branch['id'])==[{'role':'user','content':'original'},{'role':'assistant','content':'answer'}]
+    assert s.get_hermes_session(a,branch['id']) is None
+    assert s.get_conversation(a,branch['id'])['runs']==[]
+    assert all(m['run_id'] is None for m in s.get_conversation(a,branch['id'])['messages'])
+    edited = s.branch(a,cid,u['id'],content='replacement')
+    assert s.history(a,edited['id'])==[]
+    assert len(s.history(a,cid))==3
+    with pytest.raises(KeyError): s.branch(b,cid,u['id'])
+    with pytest.raises(ValueError): s.branch(a,cid,reply['id'],content='bad edit')
+    s.set_hermes_session(a,branch['id'],'branch-session')
+    with pytest.raises(ChatConflict): s.set_session(a,branch['id'],'branch-session',seeded=False)
+    with pytest.raises(ChatConflict): s.set_hermes_session(a,branch['id'],'different')
+    assert 'trusted-session' not in str(s.get_conversation(a,cid))
+
+
+def test_pagination_search_and_revocation(setup):
+    accounts,s,a,b = setup
+    c1 = s.create_conversation(a,title='Needle')['id']
+    c2 = s.create_conversation(a,title='Other')['id']
+    s.create_conversation(b,title='Needle')
+    s.add_message(a,c2,'user','searchable phrase')
+    assert len(s.list_conversations(a,q='needle')['conversations'])==1
+    assert s.list_conversations(a,q='searchable')['conversations'][0]['id']==c2
+    first = s.list_conversations(a,limit=1)
+    second = s.list_conversations(a,cursor=first['next_cursor'],limit=1)
+    assert first['conversations'][0]['id']!=second['conversations'][0]['id']
+    assert second['next_cursor'] is None
+    s.add_message(a,c1,'user','1'); s.add_message(a,c1,'assistant','2')
+    page = s.get_conversation(a,c1,limit=1)
+    assert s.get_conversation(a,c1,cursor=page['next_cursor'],limit=1)['messages'][0]['content']=='1'
+    accounts.begin_delete(b)
+    with pytest.raises(KeyError): s.get_preferences(b)
+    accounts.finish_delete(b)
+    assert accounts.db.execute('PRAGMA foreign_key_check').fetchall()==[]
+
+
+def test_disk_reopen_preserves_admission_and_events(tmp_path):
+    path=str(tmp_path/'persist.db')
+    accounts=Accounts(path)
+    owner=accounts.create('alice','Alice','admin','correct-password-alice')['id']
+    store=ChatStore(accounts)
+    cid=store.create_conversation(owner)['id']
+    admitted=store.send(owner,cid,'durable','key')
+    event=store.append_event(owner,admitted['run']['id'],'run',{'status':'queued'})
+    accounts.close()
+    accounts=Accounts(path)
+    store=ChatStore(accounts)
+    assert store.send(owner,cid,'durable','key')==admitted
+    assert store.events(owner,admitted['run']['id'])==[event]
+    accounts.close()