wip: thread revamp
This commit is contained in:
1 parent
5b3f8b0f09
commit
9277b5a770
11 files changed
+401
-205
No files matched your search
+12
-9
@@ -82,21 +82,24 @@ class Agent:
|
|||||||
return out_stream
|
return out_stream
|
||||||
|
|
||||||
async def received_message(self, message):
|
async def received_message(self, message):
|
||||||
chat_view = await self.app.get_chat_view(message.chat_id)
|
chat = await self.app.get_chat(message.chat_id)
|
||||||
thread = chat_view.get_thread(message.thread_id)
|
thread = chat.get_thread(message.thread_id)
|
||||||
|
thread.skip_list += ['log', 'resource']
|
||||||
|
await thread.load_page()
|
||||||
|
|
||||||
remotes = []
|
|
||||||
messages = []
|
messages = []
|
||||||
for msg in thread.messages:
|
for msg in thread.messages:
|
||||||
role = self._role_for_message(msg)
|
role = self._role_for_message(msg)
|
||||||
messages.append({"role": role, "content": msg.content})
|
messages.append({"role": role, "content": msg.content})
|
||||||
|
|
||||||
for msg in thread.resources:
|
remotes = []
|
||||||
# Remote declarations: content is the remote name
|
def add_remote(name):
|
||||||
if msg.subkind == 'remote':
|
if name not in remotes and (self.user_id, name) in self.app.remotes:
|
||||||
name = msg.content
|
remotes.append(name)
|
||||||
if (self.user_id, name) in self.app.remotes and name not in remotes:
|
|
||||||
remotes.append(name)
|
remote_messages = await thread.fetch_kind_page('resource', 'remote')
|
||||||
|
for msg in remote_messages:
|
||||||
|
add_remote(msg)
|
||||||
|
|
||||||
run = AgentRun.create(
|
run = AgentRun.create(
|
||||||
self.app,
|
self.app,
|
||||||
|
|||||||
@@ -19,7 +19,6 @@ import agents as agents_mod
|
|||||||
import temporary as tmp
|
import temporary as tmp
|
||||||
from server import start_server
|
from server import start_server
|
||||||
from drives import init_drives
|
from drives import init_drives
|
||||||
from chat import ChatView, Message, MESSAGE_PAGE_SIZE
|
|
||||||
from remote import RemoteConnection, RemoteToolSpec, parse_function_specs
|
from remote import RemoteConnection, RemoteToolSpec, parse_function_specs
|
||||||
|
|
||||||
from app_mixins import ChatMixin, AgentMixin
|
from app_mixins import ChatMixin, AgentMixin
|
||||||
|
|||||||
@@ -3,9 +3,20 @@ from datetime import datetime
|
|||||||
|
|
||||||
import database as db
|
import database as db
|
||||||
import temporary as tmp
|
import temporary as tmp
|
||||||
from chat import ChatView, Message, MESSAGE_PAGE_SIZE
|
from chat import Chat
|
||||||
from stream import Stream
|
from stream import Stream
|
||||||
|
|
||||||
|
def message_json(msg):
|
||||||
|
return {
|
||||||
|
"id": msg.id,
|
||||||
|
"time": msg.time,
|
||||||
|
"sender": msg.sender,
|
||||||
|
"kind": msg.kind,
|
||||||
|
"subkind": msg.subkind,
|
||||||
|
"content": msg.content,
|
||||||
|
"threadId": msg.thread_id,
|
||||||
|
}
|
||||||
|
|
||||||
class ChatMixin:
|
class ChatMixin:
|
||||||
async def create_chat(self, user_id, name, description, picture, members=[], temporary=False):
|
async def create_chat(self, user_id, name, description, picture, members=[], temporary=False):
|
||||||
if temporary:
|
if temporary:
|
||||||
@@ -30,6 +41,12 @@ class ChatMixin:
|
|||||||
|
|
||||||
return chat.id
|
return chat.id
|
||||||
|
|
||||||
|
# Gets a chat view from a given chat id
|
||||||
|
async def get_chat(self, chat_id):
|
||||||
|
chat = Chat(self, chat_id)
|
||||||
|
await chat.load()
|
||||||
|
return chat
|
||||||
|
|
||||||
async def get_chats(self, user_id):
|
async def get_chats(self, user_id):
|
||||||
# Direct filtering on the model class
|
# Direct filtering on the model class
|
||||||
chats = await db.Chat.filter(user_id=user_id).all()
|
chats = await db.Chat.filter(user_id=user_id).all()
|
||||||
@@ -74,27 +91,42 @@ class ChatMixin:
|
|||||||
|
|
||||||
# TODO: Stop repetition.
|
# TODO: Stop repetition.
|
||||||
# There's db.Message and chat.Message, also should be combined somehow, or wrap db always.
|
# There's db.Message and chat.Message, also should be combined somehow, or wrap db always.
|
||||||
return [{
|
return [message_json(msg) for msg in messages]
|
||||||
"id": message.id,
|
|
||||||
"time": message.time,
|
|
||||||
"sender": message.sender,
|
|
||||||
"kind": message.kind,
|
|
||||||
"subkind": message.subkind,
|
|
||||||
"content": message.content,
|
|
||||||
"threadId": message.thread_id,
|
|
||||||
} for message in messages]
|
|
||||||
|
|
||||||
async def create_chat_thread(self, chat_id, name=None):
|
async def get_chat_thread_heads(self, chat_id, limit_message_id=None):
|
||||||
|
chat = await self.get_chat(chat_id)
|
||||||
|
messages = await chat.load_thread_heads(limit_message_id)
|
||||||
|
return [message_json(msg) for msg in messages]
|
||||||
|
|
||||||
|
async def get_thread_messages(self, chat_id, thread_id, limit_message_id=None):
|
||||||
|
chat = await self.get_chat(chat_id)
|
||||||
|
thread = chat.get_thread(thread_id)
|
||||||
|
messages = await thread.load_page(limit_message_id)
|
||||||
|
return [message_json(msg) for msg in messages]
|
||||||
|
|
||||||
|
async def create_chat_thread(self, chat_id, parent_id=None, name=None):
|
||||||
if tmp.is_tmp_id(chat_id):
|
if tmp.is_tmp_id(chat_id):
|
||||||
thread = tmp.Thread.create(self, chat_id, name)
|
thread = tmp.Thread.create(self, chat_id, name)
|
||||||
else:
|
else:
|
||||||
thread = await db.Thread.create(
|
thread = await db.Thread.create(
|
||||||
chat_id=chat_id,
|
chat_id=chat_id,
|
||||||
|
parent_id=parent_id,
|
||||||
name=name
|
name=name
|
||||||
)
|
)
|
||||||
print(f"Created thread #{thread.id}: {thread.name}")
|
print(f"Created thread #{thread.id}: {thread.name}")
|
||||||
return thread.id
|
return thread.id
|
||||||
|
|
||||||
|
async def fork_chat_thread(self, thread_id, message_id=None, name=None):
|
||||||
|
"""Move all messages from message_id onwards into a new child thread.
|
||||||
|
If none is given, this is basically create_chat_thread with an implicit chat id."""
|
||||||
|
thread = await db.Thread.get(id=thread_id)
|
||||||
|
new_id = await self.create_chat_thread(thread.chat_id, parent_id=thread_id, name=name or thread.name)
|
||||||
|
|
||||||
|
if message_id is not None:
|
||||||
|
await db.Message.filter(thread_id=thread_id, id__gte=message_id).update(thread_id=new_id)
|
||||||
|
|
||||||
|
return new_id
|
||||||
|
|
||||||
async def add_agent_to_chat(self, chat_id, agent_id):
|
async def add_agent_to_chat(self, chat_id, agent_id):
|
||||||
if tmp.is_tmp_id(chat_id):
|
if tmp.is_tmp_id(chat_id):
|
||||||
chat = self.tmp.chats[chat_id]
|
chat = self.tmp.chats[chat_id]
|
||||||
@@ -232,11 +264,6 @@ class ChatMixin:
|
|||||||
await message.save() # Saves the specific instance
|
await message.save() # Saves the specific instance
|
||||||
print(f"Updated message #{message.id} fields: {summary}")
|
print(f"Updated message #{message.id} fields: {summary}")
|
||||||
|
|
||||||
async def get_chat_view(self, chat_id):
|
|
||||||
chat_view = ChatView(self, chat_id)
|
|
||||||
await chat_view.init_load()
|
|
||||||
return chat_view
|
|
||||||
|
|
||||||
def notify_message_to_user(self, user_id, message, update=False):
|
def notify_message_to_user(self, user_id, message, update=False):
|
||||||
if user_id in self.user_connections:
|
if user_id in self.user_connections:
|
||||||
# TODO: Server's responsibility, and Message.to_json responsibility as well
|
# TODO: Server's responsibility, and Message.to_json responsibility as well
|
||||||
|
|||||||
+188
-171
@@ -1,197 +1,214 @@
|
|||||||
|
from tortoise.expressions import Q
|
||||||
from sqlalchemy.orm import Session
|
|
||||||
from sqlalchemy.sql.expression import select
|
|
||||||
|
|
||||||
import database as db
|
import database as db
|
||||||
import temporary as tmp
|
|
||||||
|
|
||||||
MESSAGE_PAGE_SIZE = 256
|
MESSAGE_PAGE_SIZE = 256
|
||||||
|
|
||||||
class Message:
|
# In this module's context, which is viewing and fetching, a thread's messages
|
||||||
def __init__(self, id, content, sender, time=None, kind='message', subkind=None, format=None, thread_id=None):
|
# include all that match their thread ids, but also the parent's thread messages,
|
||||||
self.id = id
|
# and orphan chat messages with no thread id, always in their time order, resolving conflict by id
|
||||||
self.content = content
|
|
||||||
self.sender = sender
|
|
||||||
self.time = time
|
|
||||||
self.kind = kind
|
|
||||||
self.subkind = subkind
|
|
||||||
self.format = format
|
|
||||||
self.thread_id = thread_id
|
|
||||||
|
|
||||||
def from_db_record(record):
|
|
||||||
return Message(record.id, record.content, record.sender, record.time)
|
|
||||||
|
|
||||||
def is_from_user(self):
|
|
||||||
return self.sender == "user"
|
|
||||||
|
|
||||||
def is_from_agent(self, agent_id):
|
|
||||||
return self.sender == str(agent_id)
|
|
||||||
|
|
||||||
def to_json(self):
|
|
||||||
return {
|
|
||||||
"id": self.id,
|
|
||||||
"time": self.time,
|
|
||||||
"sender": self.sender,
|
|
||||||
"kind": self.kind,
|
|
||||||
"subkind": self.subkind,
|
|
||||||
"format": self.format,
|
|
||||||
"content": self.content,
|
|
||||||
"threadId": self.thread_id,
|
|
||||||
}
|
|
||||||
|
|
||||||
def __str__(self):
|
|
||||||
time = self.time if self.time != None else 'no-time'
|
|
||||||
content = self.content.replace("\n", "\\n")
|
|
||||||
return f"Message(#{self.id} ({time}) {self.sender}: {content})"
|
|
||||||
|
|
||||||
|
# Messages are loaded via threads
|
||||||
class Thread:
|
class Thread:
|
||||||
def __init__(self, id, chat_id, name=None, description=None):
|
def __init__(self, chat, thread_id):
|
||||||
self.id = id
|
# id can be None
|
||||||
self.chat_id = chat_id
|
self.id = thread_id
|
||||||
self.name = name
|
self.chat = chat
|
||||||
self.description = description
|
|
||||||
|
|
||||||
|
self.name = None
|
||||||
|
# Always goes earliest -> latest. No stable identity guaranteed.
|
||||||
self.messages = []
|
self.messages = []
|
||||||
self.notes = []
|
|
||||||
self.resources = []
|
|
||||||
|
|
||||||
class ChatView:
|
# List's Kind (strings) or Subkinds (tuple) to ignore in main page loading
|
||||||
def __init__(self, app, chat_id, participant_type=None, participant_id=None):
|
self.skip_list = []
|
||||||
|
|
||||||
|
# Whether the thread model has been fetched
|
||||||
|
self.loaded = False
|
||||||
|
# Whether we have read all thread messages in the db
|
||||||
|
self.complete = False
|
||||||
|
|
||||||
|
# Load state
|
||||||
|
self._earliest_message_id = None
|
||||||
|
self._parent_id = None
|
||||||
|
|
||||||
|
async def load(self):
|
||||||
|
# Fetches the thread model, resolving the thread's name and parent
|
||||||
|
if self.loaded:
|
||||||
|
return
|
||||||
|
if self.id is not None:
|
||||||
|
model = await db.Thread.get_or_none(id=self.id, chat_id=self.chat.id)
|
||||||
|
if model is not None:
|
||||||
|
self.name = model.name
|
||||||
|
self._parent_id = model.parent_id
|
||||||
|
self.loaded = True
|
||||||
|
|
||||||
|
def _scope_query(self):
|
||||||
|
# Every chat message this thread's view is made of: the ones bound to
|
||||||
|
# its thread id, to its parent's thread id, or to no thread at all
|
||||||
|
query = db.Message.filter(chat_id=self.chat.id)
|
||||||
|
if self.id is None:
|
||||||
|
return query.filter(thread_id__isnull=True)
|
||||||
|
thread_ids = [self.id]
|
||||||
|
if self._parent_id is not None:
|
||||||
|
thread_ids.append(self._parent_id)
|
||||||
|
return query.filter(Q(thread_id__in=thread_ids) | Q(thread_id__isnull=True))
|
||||||
|
|
||||||
|
def _ignored(self, message):
|
||||||
|
# Whether the message matches a kind (string) or kind/subkind (tuple) skip entry
|
||||||
|
for entry in self.skip_list:
|
||||||
|
if isinstance(entry, tuple):
|
||||||
|
if message.kind == entry[0] and message.subkind == entry[1]:
|
||||||
|
return True
|
||||||
|
elif message.kind == entry:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _order_key(message):
|
||||||
|
return (message.time, message.id)
|
||||||
|
|
||||||
|
def _merge_messages(self, batch):
|
||||||
|
# Merges an earliest -> latest batch into messages, dropping already
|
||||||
|
# loaded ones and keeping the overall time order
|
||||||
|
known = {message.id for message in self.messages}
|
||||||
|
additions = [message for message in batch if message.id not in known]
|
||||||
|
if not additions:
|
||||||
|
return
|
||||||
|
if not self.messages or self._order_key(additions[-1]) < self._order_key(self.messages[0]):
|
||||||
|
self.messages = additions + self.messages
|
||||||
|
else:
|
||||||
|
self.messages = sorted(self.messages + additions, key=self._order_key)
|
||||||
|
|
||||||
|
async def load_page(self, limit_message_id=None):
|
||||||
|
# Loads the next MESSAGE_PAGE_SIZE into messages, walking back in time
|
||||||
|
# from limit_message_id or, when it is not given, from the earliest
|
||||||
|
# loaded message; only own paging updates the load state
|
||||||
|
await self.load()
|
||||||
|
own_page = limit_message_id is None
|
||||||
|
cursor = self._earliest_message_id if own_page else limit_message_id
|
||||||
|
page, cursor, exhausted = await self.chat.fetch_page(
|
||||||
|
self._scope_query(), cursor, lambda message: not self._ignored(message)
|
||||||
|
)
|
||||||
|
self._merge_messages(page)
|
||||||
|
if own_page:
|
||||||
|
self._earliest_message_id = cursor
|
||||||
|
if exhausted:
|
||||||
|
self.complete = True
|
||||||
|
return page
|
||||||
|
|
||||||
|
async def fetch_kind_page(self, kind, subkind=None, limit_message_id=None):
|
||||||
|
# Returns up to MESSAGE_PAGE_SIZE messages of the given kind (and
|
||||||
|
# subkind): the matching ones already loaded are kept for the response,
|
||||||
|
# the rest of the page is fetched from the db, offset by the count of
|
||||||
|
# the kept ones; the load state is left untouched
|
||||||
|
await self.load()
|
||||||
|
kept = [
|
||||||
|
message
|
||||||
|
for message in self.messages
|
||||||
|
if message.kind == kind
|
||||||
|
and (subkind is None or message.subkind == subkind)
|
||||||
|
and (limit_message_id is None or message.id < limit_message_id)
|
||||||
|
]
|
||||||
|
skip_count = len(kept)
|
||||||
|
fetched = []
|
||||||
|
if skip_count < MESSAGE_PAGE_SIZE:
|
||||||
|
query = self._scope_query().filter(kind=kind)
|
||||||
|
if subkind is not None:
|
||||||
|
query = query.filter(subkind=subkind)
|
||||||
|
if limit_message_id is not None:
|
||||||
|
query = query.filter(id__lt=limit_message_id)
|
||||||
|
fetched = await self.chat.fetch_latest(
|
||||||
|
query, skip_count, MESSAGE_PAGE_SIZE - skip_count
|
||||||
|
)
|
||||||
|
self._merge_messages([message for message in fetched if not self._ignored(message)])
|
||||||
|
return fetched + kept
|
||||||
|
|
||||||
|
|
||||||
|
# Chat mainly provides info, threads, and some query operations,
|
||||||
|
# message loading should be done using `Thread`
|
||||||
|
class Chat:
|
||||||
|
def __init__(self, app, chat_id):
|
||||||
|
self.id = chat_id
|
||||||
self.app = app
|
self.app = app
|
||||||
self.chat_id = chat_id
|
self.chat_id = chat_id
|
||||||
|
|
||||||
|
self.loaded = False
|
||||||
|
|
||||||
self.name = None
|
self.name = None
|
||||||
self.description = None
|
self.description = None
|
||||||
|
|
||||||
self.participant_type = participant_type
|
# A limited set of the messages for tracking chat info
|
||||||
self.participant_id = participant_id
|
self.messages = None
|
||||||
|
|
||||||
# Thread holding messages without a set thread id
|
async def load(self):
|
||||||
self.default_thread = Thread(None, chat_id)
|
# Fetches the chat's own information
|
||||||
# Most recent thread objects, earliest to latest
|
if self.loaded:
|
||||||
self.threads = []
|
return
|
||||||
# Map of threads by their id
|
chat = await db.Chat.get_or_none(id=self.id)
|
||||||
self.thread_map = {}
|
if chat is not None:
|
||||||
# Long term summary of this chat
|
self.name = chat.name
|
||||||
self.summary = None
|
self.description = chat.description
|
||||||
# Separate list for note messages from any thread
|
self.loaded = True
|
||||||
# Is it better to exclude them or include them from threads? atm they're included
|
|
||||||
self.notes = []
|
|
||||||
|
|
||||||
# Id of the thread currently in focus, by default the last thread appearing in the conversation
|
async def load_thread_heads(self, limit_message_id=None):
|
||||||
self.current_thread_id = None
|
if limit_message_id != None:
|
||||||
|
raise Exception("Chat.load_thread_heads: limit_message_id not yet supported")
|
||||||
|
|
||||||
self._earliest_message_id = None
|
# Fetches MESSAGE_PAGE_SIZE of the latest messages with unique threads
|
||||||
self._earliest_message_time = None
|
# returns the fetched list as well as storing it in self.messages
|
||||||
|
seen = set()
|
||||||
|
|
||||||
async def init_load(self):
|
def accept(message):
|
||||||
await self._load_basic_data()
|
# The walk goes latest -> earliest, so the first message met of a
|
||||||
await self._load_next_messages(MESSAGE_PAGE_SIZE)
|
# thread (the null thread of orphan messages included) is that
|
||||||
|
# thread's latest, and the only one kept per thread
|
||||||
|
if message.thread_id in seen:
|
||||||
|
return False
|
||||||
|
seen.add(message.thread_id)
|
||||||
|
return True
|
||||||
|
|
||||||
async def _load_basic_data(self):
|
heads, _, _ = await self.fetch_page(
|
||||||
if tmp.is_tmp_id(self.chat_id):
|
db.Message.filter(chat_id=self.id), None, accept
|
||||||
record = self.app.tmp.chats.get(self.chat_id)
|
)
|
||||||
else:
|
self.messages = heads
|
||||||
# Using get_or_none for cleaner existence checks
|
return heads
|
||||||
record = await db.Chat.get_or_none(id=self.chat_id)
|
|
||||||
|
|
||||||
if not record:
|
@staticmethod
|
||||||
raise Exception(f"Chat not found with id {self.chat_id}")
|
async def fetch_page(query, cursor=None, accept=None):
|
||||||
|
# Fetches up to MESSAGE_PAGE_SIZE messages by walking back in time from
|
||||||
|
# cursor (an exclusive message id, or the latest message when not
|
||||||
|
# given), keeping the ones `accept` approves (all of them when not
|
||||||
|
# given); returns the page in earliest -> latest order, the cursor to
|
||||||
|
# continue the walk from, and whether the query ran out of messages
|
||||||
|
page = []
|
||||||
|
exhausted = False
|
||||||
|
while len(page) < MESSAGE_PAGE_SIZE:
|
||||||
|
batch_query = query
|
||||||
|
if cursor is not None:
|
||||||
|
batch_query = batch_query.filter(id__lt=cursor)
|
||||||
|
rows = await batch_query.order_by("-time", "-id").limit(MESSAGE_PAGE_SIZE)
|
||||||
|
if not rows:
|
||||||
|
exhausted = True
|
||||||
|
break
|
||||||
|
for row in rows:
|
||||||
|
cursor = row.id
|
||||||
|
if accept is None or accept(row):
|
||||||
|
page.append(row)
|
||||||
|
if len(page) == MESSAGE_PAGE_SIZE:
|
||||||
|
break
|
||||||
|
if len(page) < MESSAGE_PAGE_SIZE and len(rows) < MESSAGE_PAGE_SIZE:
|
||||||
|
exhausted = True
|
||||||
|
break
|
||||||
|
page.reverse()
|
||||||
|
return page, cursor, exhausted
|
||||||
|
|
||||||
self.name = record.name
|
@staticmethod
|
||||||
self.description = record.description
|
async def fetch_latest(query, offset=0, limit=MESSAGE_PAGE_SIZE):
|
||||||
|
# Fetches up to `limit` of the newest messages in the query, skipping
|
||||||
async def _load_next_messages(self, load_count):
|
# the newest `offset` ones, in earliest -> latest order
|
||||||
if tmp.is_tmp_id(self.chat_id):
|
rows = await query.offset(offset).order_by("-time", "-id").limit(limit)
|
||||||
chat = self.app.tmp.chats[self.chat_id]
|
return list(reversed(rows))
|
||||||
record_batch = [
|
|
||||||
msg for msg in chat.messages
|
|
||||||
if not self._earliest_message_time or msg.id < self._earliest_message_time
|
|
||||||
][:load_count]
|
|
||||||
record_batch.reverse()
|
|
||||||
records_rev = record_batch
|
|
||||||
else:
|
|
||||||
# Start building the filter
|
|
||||||
filters = {"chat_id": self.chat_id}
|
|
||||||
|
|
||||||
# Add the 'earliest message' condition if it exists
|
|
||||||
if self._earliest_message_id is not None:
|
|
||||||
filters["id__lte"] = self._earliest_message_id
|
|
||||||
|
|
||||||
# Tortoise query: filter, order, and limit
|
|
||||||
# ordering by id is more robust than by time
|
|
||||||
records_rev = await db.Message.filter(**filters).order_by("-id").limit(load_count)
|
|
||||||
|
|
||||||
# Update the earliest id for the next load operation
|
|
||||||
if records_rev:
|
|
||||||
# Note: Since we ordered by -time (descending), index 0 is the newest in this batch (highest id/time)
|
|
||||||
# The "earliest" or oldest is the one with the smallest ID/Time in this set
|
|
||||||
self._earliest_message_id = records_rev[-1].id
|
|
||||||
self._earliest_message_time = records_rev[-1].time
|
|
||||||
|
|
||||||
if not self.threads:
|
|
||||||
# current thread is the latest known from the newest record in batch
|
|
||||||
self.current_thread_id = records_rev[0].thread_id
|
|
||||||
|
|
||||||
# Go over messages from newest to oldest
|
|
||||||
for record in records_rev:
|
|
||||||
if record.thread_id is None:
|
|
||||||
thread = self.default_thread
|
|
||||||
else:
|
|
||||||
thread = self.thread_map.get(record.thread_id)
|
|
||||||
if not thread:
|
|
||||||
# Assuming Thread is a local class, not the DB model
|
|
||||||
thread = Thread(record.thread_id, self.chat_id)
|
|
||||||
self.thread_map[thread.id] = thread
|
|
||||||
self.threads.insert(0, thread)
|
|
||||||
|
|
||||||
# Create local Message object
|
|
||||||
message = Message(
|
|
||||||
record.id,
|
|
||||||
record.content,
|
|
||||||
record.sender,
|
|
||||||
record.time,
|
|
||||||
record.kind,
|
|
||||||
record.subkind
|
|
||||||
)
|
|
||||||
|
|
||||||
if message.kind == 'note':
|
|
||||||
thread.notes.insert(0, message)
|
|
||||||
elif message.kind == 'resource':
|
|
||||||
thread.resources.insert(0, message)
|
|
||||||
else:
|
|
||||||
thread.messages.insert(0, message)
|
|
||||||
|
|
||||||
def get_thread(self, thread_id):
|
def get_thread(self, thread_id):
|
||||||
if thread_id == None:
|
return Thread(self, thread_id)
|
||||||
return self.default_thread
|
|
||||||
return self.thread_map.get(thread_id)
|
|
||||||
|
|
||||||
def get_current_thread(self):
|
|
||||||
# TODO: if !thread
|
|
||||||
return self.get_thread(self.current_thread_id)
|
|
||||||
|
|
||||||
def debug_string(self):
|
|
||||||
result_lines = []
|
|
||||||
def _print(line):
|
|
||||||
result_lines.append(line)
|
|
||||||
|
|
||||||
def print_thread(thread, title=None):
|
|
||||||
if title:
|
|
||||||
_print(title + ":")
|
|
||||||
elif thread.name:
|
|
||||||
_print(thread.name + ":")
|
|
||||||
else:
|
|
||||||
_print(f"Thread {thread.id}:")
|
|
||||||
for message in thread.messages:
|
|
||||||
_print(" " + str(message))
|
|
||||||
|
|
||||||
print_thread(self.default_thread, "Default Thread")
|
|
||||||
for thread in self.threads:
|
|
||||||
print_thread(thread)
|
|
||||||
|
|
||||||
_print("Current thread id: " + str(self.current_thread_id))
|
|
||||||
|
|
||||||
return "\n".join(result_lines)
|
|
||||||
|
|
||||||
|
|
||||||
@@ -123,9 +123,18 @@ class Thread(models.Model):
|
|||||||
id = fields.IntField(pk=True)
|
id = fields.IntField(pk=True)
|
||||||
chat = fields.ForeignKeyField("models.Chat", related_name="threads")
|
chat = fields.ForeignKeyField("models.Chat", related_name="threads")
|
||||||
|
|
||||||
|
parent = fields.ForeignKeyField("models.Thread", related_name="children", null=True)
|
||||||
|
|
||||||
|
# Optional Description fields
|
||||||
name = fields.CharField(max_length=255, null=True)
|
name = fields.CharField(max_length=255, null=True)
|
||||||
description = fields.TextField(null=True)
|
description = fields.TextField(null=True)
|
||||||
|
|
||||||
|
# Automatic summary fields, include parent messages
|
||||||
|
first_message = fields.ForeignKeyField("models.Message", related_name=False, null=True, on_delete=fields.SET_NULL)
|
||||||
|
last_message = fields.ForeignKeyField("models.Message", related_name=False, null=True, on_delete=fields.SET_NULL)
|
||||||
|
# null make life easy
|
||||||
|
message_count = fields.IntField(default=0)
|
||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
table = "threads"
|
table = "threads"
|
||||||
|
|
||||||
@@ -160,6 +169,7 @@ class MessageDatum(models.Model):
|
|||||||
class Meta:
|
class Meta:
|
||||||
table = "message_data"
|
table = "message_data"
|
||||||
|
|
||||||
|
# TODO: deprecate, manage models with resources
|
||||||
class Model(models.Model):
|
class Model(models.Model):
|
||||||
id = fields.IntField(pk=True)
|
id = fields.IntField(pk=True)
|
||||||
#user = fields.ForeignKeyField("models.User", null=True)
|
#user = fields.ForeignKeyField("models.User", null=True)
|
||||||
|
|||||||
@@ -0,0 +1,17 @@
|
|||||||
|
from tortoise import migrations
|
||||||
|
from tortoise.migrations import operations as ops
|
||||||
|
from tortoise.fields.base import OnDelete
|
||||||
|
from tortoise import fields
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
dependencies = [('models', '0002_auto_20260921_1840')]
|
||||||
|
|
||||||
|
initial = False
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
ops.AddField(
|
||||||
|
model_name='Thread',
|
||||||
|
name='parent',
|
||||||
|
field=fields.ForeignKeyField('models.Thread', source_field='parent_id', null=True, db_constraint=True, to_field='id', related_name='children', on_delete=OnDelete.CASCADE),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -0,0 +1,35 @@
|
|||||||
|
from tortoise import migrations
|
||||||
|
from tortoise.migrations import operations as ops
|
||||||
|
from tortoise.fields.base import OnDelete
|
||||||
|
from tortoise import fields
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
dependencies = [('models', '0003_thread_parent')]
|
||||||
|
|
||||||
|
initial = False
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
ops.AddField(
|
||||||
|
model_name='Thread',
|
||||||
|
name='first_message',
|
||||||
|
field=fields.ForeignKeyField('models.Message', source_field='first_message_id', null=True, db_constraint=True, to_field='id', related_name=False, on_delete=OnDelete.SET_NULL),
|
||||||
|
),
|
||||||
|
ops.AddField(
|
||||||
|
model_name='Thread',
|
||||||
|
name='last_message',
|
||||||
|
field=fields.ForeignKeyField('models.Message', source_field='last_message_id', null=True, db_constraint=True, to_field='id', related_name=False, on_delete=OnDelete.SET_NULL),
|
||||||
|
),
|
||||||
|
ops.AddField(
|
||||||
|
model_name='Thread',
|
||||||
|
name='message_count',
|
||||||
|
# Null true to allow inserting in sqlite (annoying ass bug)
|
||||||
|
field=fields.IntField(null=True, default=0),
|
||||||
|
),
|
||||||
|
# Fix sqlite insert
|
||||||
|
ops.AlterField(
|
||||||
|
model_name='Thread',
|
||||||
|
name='message_count',
|
||||||
|
# Now null is false
|
||||||
|
field=fields.IntField(null=False, default=0),
|
||||||
|
),
|
||||||
|
]
|
||||||
@@ -234,6 +234,21 @@ def register(server, app, auth_manager):
|
|||||||
)
|
)
|
||||||
await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
|
await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
|
||||||
|
|
||||||
|
elif action == "get-chat-thread-heads":
|
||||||
|
messages_data = await app.get_chat_thread_heads(
|
||||||
|
body['chatId'],
|
||||||
|
limit_message_id=body.get('limitMessageId', None)
|
||||||
|
)
|
||||||
|
await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
|
||||||
|
|
||||||
|
elif action == "get-thread-messages":
|
||||||
|
messages_data = await app.get_thread_messages(
|
||||||
|
body['chatId'],
|
||||||
|
body['threadId'],
|
||||||
|
limit_message_id=body.get('limitMessageId', None)
|
||||||
|
)
|
||||||
|
await send({'message': "chat messages retrieved", 'messages': messages_data, 'nonce': nonce})
|
||||||
|
|
||||||
elif action == "send-chat-message":
|
elif action == "send-chat-message":
|
||||||
message = await app.send_chat_message(
|
message = await app.send_chat_message(
|
||||||
chat_id=body['chatId'],
|
chat_id=body['chatId'],
|
||||||
@@ -262,6 +277,13 @@ def register(server, app, auth_manager):
|
|||||||
)
|
)
|
||||||
await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
|
await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
|
||||||
|
|
||||||
|
elif action == "fork-chat-thread":
|
||||||
|
thread_id = await app.fork_chat_thread(
|
||||||
|
thread_id=body['threadId'],
|
||||||
|
message_id=body['messageId'],
|
||||||
|
)
|
||||||
|
await send({'message': "chat thread created", 'threadId': thread_id, 'nonce': nonce})
|
||||||
|
|
||||||
elif action == "send-agent-prompt":
|
elif action == "send-agent-prompt":
|
||||||
if 'instruction' not in body and 'input' not in body:
|
if 'instruction' not in body and 'input' not in body:
|
||||||
await send({'error': "instruction or input required"})
|
await send({'error': "instruction or input required"})
|
||||||
|
|||||||
+64
-3
@@ -79,6 +79,8 @@ export class ChatSession {
|
|||||||
this._rescroll = callback;
|
this._rescroll = callback;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Call after changing the messages display length (change list or content)
|
||||||
|
// It protects the scroll position in the document
|
||||||
rescroll (options) {
|
rescroll (options) {
|
||||||
if (this._rescroll) this._rescroll(options);
|
if (this._rescroll) this._rescroll(options);
|
||||||
}
|
}
|
||||||
@@ -117,7 +119,9 @@ export class ChatSession {
|
|||||||
this._unsubscribeChat = backend.onChatMessages(chatId, it => this.receiveMessage(it));
|
this._unsubscribeChat = backend.onChatMessages(chatId, it => this.receiveMessage(it));
|
||||||
this._unsubscribeRuns = runs.onChatMessage(chatId, event => this.trackRun(event));
|
this._unsubscribeRuns = runs.onChatMessage(chatId, event => this.trackRun(event));
|
||||||
|
|
||||||
const response = await backend.sendAction("get-chat-messages", {chatId});
|
// Don't load all messages, only thread headers
|
||||||
|
//const response = await backend.sendAction("get-chat-messages", {chatId});
|
||||||
|
const response = await backend.sendAction("get-chat-thread-heads", {chatId});
|
||||||
if (this.chatId != chatId) return;
|
if (this.chatId != chatId) return;
|
||||||
|
|
||||||
// Always good to duplicate when assigning to a ref
|
// Always good to duplicate when assigning to a ref
|
||||||
@@ -134,8 +138,7 @@ export class ChatSession {
|
|||||||
// otherwise start in "New Thread" state
|
// otherwise start in "New Thread" state
|
||||||
const lastMessage = response.messages[response.messages.length - 1];
|
const lastMessage = response.messages[response.messages.length - 1];
|
||||||
if (lastMessage && lastMessage.threadId != null && !isMessageStale(lastMessage)) {
|
if (lastMessage && lastMessage.threadId != null && !isMessageStale(lastMessage)) {
|
||||||
this.currentThreadId.value = lastMessage.threadId;
|
this.activateThread(lastMessage.threadId);
|
||||||
this.isNewThread.value = false;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Must happen on the same async-thread that sets `messages`
|
// Must happen on the same async-thread that sets `messages`
|
||||||
@@ -317,6 +320,21 @@ export class ChatSession {
|
|||||||
// Continuing a thread: it becomes the one open thread
|
// Continuing a thread: it becomes the one open thread
|
||||||
this.currentThreadId.value = threadId;
|
this.currentThreadId.value = threadId;
|
||||||
this.isNewThread.value = false;
|
this.isNewThread.value = false;
|
||||||
|
|
||||||
|
// Load thread messages asynchronously
|
||||||
|
backend.sendAction("get-thread-messages", {
|
||||||
|
chatId: this.chatId, threadId
|
||||||
|
}).then(res => {
|
||||||
|
this.addMessageBulk(res.messages.map(it => ({
|
||||||
|
id: it.id,
|
||||||
|
type: it.kind,
|
||||||
|
format: it.format,
|
||||||
|
subkind: it.subkind,
|
||||||
|
sender: it.sender,
|
||||||
|
content: it.content,
|
||||||
|
threadId: it.threadId,
|
||||||
|
})));
|
||||||
|
});
|
||||||
this.rescroll();
|
this.rescroll();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -327,6 +345,49 @@ export class ChatSession {
|
|||||||
this.rescroll();
|
this.rescroll();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Adds a bulk of messages to the message list, in their correct
|
||||||
|
// chronological position and replacing duplicates
|
||||||
|
addMessageBulk (messages) {
|
||||||
|
const list = this.messages.value;
|
||||||
|
|
||||||
|
for (let message of messages) {
|
||||||
|
const id = message.id;
|
||||||
|
|
||||||
|
// Replace an already-known message in place
|
||||||
|
const existingIndex = list.findIndex(it => it.id == id);
|
||||||
|
if (existingIndex >= 0) {
|
||||||
|
list[existingIndex] = message;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// An unsent user message echoed back with an id: adopt it
|
||||||
|
if (message.sender == 'user') {
|
||||||
|
const unsentIndex = this.unsentMessages.value.findIndex(
|
||||||
|
it => it.content == message.content
|
||||||
|
);
|
||||||
|
if (unsentIndex >= 0) {
|
||||||
|
this.unsentMessages.value.splice(unsentIndex, 1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Insert after the last message known to belong to the same thread,
|
||||||
|
// keeping each thread's messages contiguous and ordered
|
||||||
|
let insertIndex = -1;
|
||||||
|
for (let i = list.length - 1; i >= 0; i--) {
|
||||||
|
if (list[i].threadId != null && list[i].threadId == message.threadId) {
|
||||||
|
insertIndex = i + 1;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (insertIndex < 0) insertIndex = list.length;
|
||||||
|
list.splice(insertIndex, 0, message);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Refs need reassignment to notify watchers; duplicate the list
|
||||||
|
this.messages.value = list.slice();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
// === Entry Computation === //
|
// === Entry Computation === //
|
||||||
|
|
||||||
|
|||||||
@@ -114,8 +114,9 @@
|
|||||||
async function chatEntryContextMenu (message, logMeta) {
|
async function chatEntryContextMenu (message, logMeta) {
|
||||||
let entries = [
|
let entries = [
|
||||||
{value: 'pin', title: "Pin message"},
|
{value: 'pin', title: "Pin message"},
|
||||||
{value: 'edit', title: "Edit message"},
|
{value: 'edit', title: "Edit in-place"},
|
||||||
{value: 'start-thread', title: "Start thread from here"},
|
{value: 'edit-fork', title: "Edit in new thread"},
|
||||||
|
//{value: 'start-thread', title: "Start thread from here"},
|
||||||
];
|
];
|
||||||
|
|
||||||
// Long logs can be expanded/collapsed
|
// Long logs can be expanded/collapsed
|
||||||
@@ -136,8 +137,8 @@
|
|||||||
editMessage(message);
|
editMessage(message);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case 'start-thread': {
|
case 'edit-fork': {
|
||||||
alert("Start Thread (TODO)");
|
editMessage(message, true);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
case 'toggle-log': {
|
case 'toggle-log': {
|
||||||
|
|||||||
@@ -53,7 +53,11 @@ case "$1" in
|
|||||||
# Create migrations based on the current schema in database.py
|
# Create migrations based on the current schema in database.py
|
||||||
"migration")
|
"migration")
|
||||||
cd backend/src
|
cd backend/src
|
||||||
tortoise -c database.TORTOISE_ORM makemigrations
|
if [ -z "$2" ]; then
|
||||||
|
tortoise -c database.TORTOISE_ORM makemigrations
|
||||||
|
else
|
||||||
|
tortoise -c database.TORTOISE_ORM makemigrations --name "$2"
|
||||||
|
fi
|
||||||
;;
|
;;
|
||||||
|
|
||||||
"deploy")
|
"deploy")
|
||||||
|
|||||||
Reference in new issue
Block a user