478 lines
31 KiB
Python
478 lines
31 KiB
Python
"""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
|