From bfbaef8c687502f9626c1be2ba31b5be9ea1324f Mon Sep 17 00:00:00 2001 From: Shane Maynard Date: Sat, 3 Oct 2026 08:51:09 -0400 Subject: [PATCH] feat: add durable owner-scoped chat store --- control/chat_store.py | 468 ++++++++++++++++++++++++++++++++++ tests/unit/test_chat_store.py | 161 ++++++++++++ 2 files changed, 629 insertions(+) create mode 100644 control/chat_store.py create mode 100644 tests/unit/test_chat_store.py diff --git a/control/chat_store.py b/control/chat_store.py new file mode 100644 index 0000000..e63f2fd --- /dev/null +++ b/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 positionlimit 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 diff --git a/tests/unit/test_chat_store.py b/tests/unit/test_chat_store.py new file mode 100644 index 0000000..fcce681 --- /dev/null +++ b/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()