bvandeusen
d352e9264b
Also rename services/tz.user_briefing_date → user_day_date with a backwards compat alias (briefing modules using the old name will be deleted in the upcoming briefing tear-down stage). Update services/chat.py to_dict to use day_date. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
296 lines
10 KiB
Python
296 lines
10 KiB
Python
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import func, select, delete as sa_delete
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.conversation import Conversation, Message
|
|
from fabledassistant.config import Config
|
|
from fabledassistant.services.llm import generate_completion
|
|
from fabledassistant.services.notes import create_note
|
|
from fabledassistant.services.settings import get_setting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def create_conversation(
|
|
user_id: int, title: str = "", model: str = "", conversation_type: str = "chat"
|
|
) -> Conversation:
|
|
async with async_session() as session:
|
|
conv = Conversation(user_id=user_id, title=title, model=model, conversation_type=conversation_type)
|
|
session.add(conv)
|
|
await session.commit()
|
|
# Re-fetch with messages eagerly loaded to avoid lazy-load in async context
|
|
result = await session.execute(
|
|
select(Conversation)
|
|
.options(selectinload(Conversation.messages))
|
|
.where(Conversation.id == conv.id)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
|
|
async def get_conversation(
|
|
user_id: int, conversation_id: int
|
|
) -> Conversation | None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Conversation)
|
|
.options(selectinload(Conversation.messages))
|
|
.where(
|
|
Conversation.id == conversation_id,
|
|
Conversation.user_id == user_id,
|
|
)
|
|
)
|
|
return result.scalars().first()
|
|
|
|
|
|
async def list_conversations(
|
|
user_id: int, limit: int = 50, offset: int = 0, conv_type: str = "chat"
|
|
) -> tuple[list[dict], int]:
|
|
async with async_session() as session:
|
|
total = await session.scalar(
|
|
select(func.count(Conversation.id)).where(
|
|
Conversation.user_id == user_id,
|
|
Conversation.conversation_type == conv_type,
|
|
)
|
|
) or 0
|
|
|
|
# Subquery for message count — avoids loading all messages
|
|
msg_count = (
|
|
select(func.count(Message.id))
|
|
.where(Message.conversation_id == Conversation.id)
|
|
.correlate(Conversation)
|
|
.scalar_subquery()
|
|
)
|
|
|
|
result = await session.execute(
|
|
select(Conversation, msg_count.label("message_count"))
|
|
.where(Conversation.user_id == user_id, Conversation.conversation_type == conv_type)
|
|
.order_by(Conversation.updated_at.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
|
|
conversations = []
|
|
for row in result.all():
|
|
conv = row[0]
|
|
d = {
|
|
"id": conv.id,
|
|
"title": conv.title,
|
|
"model": conv.model,
|
|
"conversation_type": conv.conversation_type,
|
|
"day_date": conv.day_date.isoformat() if conv.day_date else None,
|
|
"rag_project_id": conv.rag_project_id,
|
|
"message_count": row[1],
|
|
"created_at": conv.created_at.isoformat(),
|
|
"updated_at": conv.updated_at.isoformat(),
|
|
}
|
|
conversations.append(d)
|
|
|
|
return conversations, total
|
|
|
|
|
|
async def delete_conversation(user_id: int, conversation_id: int) -> bool:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Conversation).where(
|
|
Conversation.id == conversation_id,
|
|
Conversation.user_id == user_id,
|
|
)
|
|
)
|
|
conv = result.scalars().first()
|
|
if conv is None:
|
|
return False
|
|
await session.delete(conv)
|
|
await session.commit()
|
|
return True
|
|
|
|
|
|
async def bulk_delete_conversations(user_id: int, ids: list[int]) -> int:
|
|
"""Delete multiple conversations by ID for a user. Returns count deleted."""
|
|
if not ids:
|
|
return 0
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
sa_delete(Conversation)
|
|
.where(Conversation.user_id == user_id, Conversation.id.in_(ids))
|
|
.returning(Conversation.id)
|
|
)
|
|
await session.commit()
|
|
return len(result.fetchall())
|
|
|
|
|
|
async def cleanup_old_conversations(user_id: int, days: int) -> int:
|
|
"""Delete conversations older than `days` days. Returns count deleted."""
|
|
if days <= 0:
|
|
return 0
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
sa_delete(Conversation)
|
|
.where(
|
|
Conversation.user_id == user_id,
|
|
Conversation.updated_at < cutoff,
|
|
Conversation.conversation_type != "mcp", # preserve MCP audit trail
|
|
Conversation.conversation_type != "voice", # voice convs managed separately
|
|
Conversation.conversation_type != "briefing", # briefing history managed by briefing system
|
|
)
|
|
.returning(Conversation.id)
|
|
)
|
|
await session.commit()
|
|
return len(result.fetchall())
|
|
|
|
|
|
_UNSET = object()
|
|
|
|
|
|
async def update_conversation(
|
|
user_id: int,
|
|
conversation_id: int,
|
|
title: str | None = None,
|
|
model: str | None = None,
|
|
rag_project_id: object = _UNSET,
|
|
) -> Conversation | None:
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Conversation).where(
|
|
Conversation.id == conversation_id,
|
|
Conversation.user_id == user_id,
|
|
)
|
|
)
|
|
conv = result.scalars().first()
|
|
if conv is None:
|
|
return None
|
|
if title is not None:
|
|
conv.title = title
|
|
if model is not None:
|
|
conv.model = model
|
|
if rag_project_id is not _UNSET:
|
|
conv.rag_project_id = rag_project_id # type: ignore[assignment]
|
|
conv.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
await session.refresh(conv)
|
|
return conv
|
|
|
|
|
|
async def update_conversation_title(
|
|
user_id: int, conversation_id: int, title: str
|
|
) -> Conversation | None:
|
|
return await update_conversation(user_id, conversation_id, title=title)
|
|
|
|
|
|
async def add_message(
|
|
conversation_id: int,
|
|
role: str,
|
|
content: str,
|
|
context_note_id: int | None = None,
|
|
status: str | None = None,
|
|
tool_calls: list | None = None,
|
|
msg_metadata: dict | None = None,
|
|
) -> Message:
|
|
async with async_session() as session:
|
|
kwargs: dict = dict(
|
|
conversation_id=conversation_id,
|
|
role=role,
|
|
content=content,
|
|
context_note_id=context_note_id,
|
|
)
|
|
if status is not None:
|
|
kwargs["status"] = status
|
|
if tool_calls is not None:
|
|
kwargs["tool_calls"] = tool_calls
|
|
if msg_metadata is not None:
|
|
kwargs["msg_metadata"] = msg_metadata
|
|
msg = Message(**kwargs)
|
|
session.add(msg)
|
|
# Touch conversation updated_at
|
|
conv = await session.get(Conversation, conversation_id)
|
|
if conv:
|
|
conv.updated_at = datetime.now(timezone.utc)
|
|
await session.commit()
|
|
await session.refresh(msg)
|
|
return msg
|
|
|
|
|
|
async def get_message(message_id: int) -> Message | None:
|
|
async with async_session() as session:
|
|
return await session.get(Message, message_id)
|
|
|
|
|
|
async def save_response_as_note(user_id: int, message_id: int) -> dict:
|
|
"""Create a note from an assistant message. Returns the new note dict."""
|
|
msg = await get_message(message_id)
|
|
if msg is None:
|
|
raise ValueError("Message not found")
|
|
if msg.role != "assistant":
|
|
raise ValueError("Can only save assistant messages as notes")
|
|
|
|
conv = await get_conversation(user_id, msg.conversation_id)
|
|
|
|
# Generate title via LLM using the assistant message content
|
|
title = ""
|
|
if conv:
|
|
try:
|
|
prompt_messages = [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"Generate a concise 3-8 word title for a note based on "
|
|
"this content. Reply with ONLY the title, no quotes or "
|
|
"punctuation."
|
|
),
|
|
},
|
|
{"role": "user", "content": msg.content[:2000]},
|
|
]
|
|
bg_model = await get_setting(user_id, "background_model", Config.OLLAMA_BACKGROUND_MODEL)
|
|
title = await generate_completion(prompt_messages, bg_model)
|
|
title = title.strip().strip('"\'').strip()[:100]
|
|
except Exception:
|
|
logger.warning("Failed to generate note title, using fallback", exc_info=True)
|
|
|
|
if not title:
|
|
lines = msg.content.strip().split("\n", 1)
|
|
title = lines[0].strip().lstrip("# ")[:100]
|
|
|
|
note = await create_note(user_id, title=title, body=msg.content, tags=["chat"])
|
|
return note.to_dict()
|
|
|
|
|
|
async def summarize_conversation_as_note(
|
|
user_id: int, conversation_id: int, model: str
|
|
) -> dict:
|
|
"""Summarize a conversation using the LLM and save as a note."""
|
|
conv = await get_conversation(user_id, conversation_id)
|
|
if conv is None:
|
|
raise ValueError("Conversation not found")
|
|
|
|
# Build the conversation text
|
|
conv_text = []
|
|
for msg in conv.messages:
|
|
if msg.role == "system":
|
|
continue
|
|
label = "User" if msg.role == "user" else "Assistant"
|
|
conv_text.append(f"{label}: {msg.content}")
|
|
|
|
prompt_messages = [
|
|
{
|
|
"role": "system",
|
|
"content": (
|
|
"Summarize the following conversation into a concise note. "
|
|
"Include key points, decisions, and any action items. "
|
|
"Format the summary in markdown."
|
|
),
|
|
},
|
|
{"role": "user", "content": "\n\n".join(conv_text)},
|
|
]
|
|
|
|
logger.info("Summarizing conversation %d with model %s", conversation_id, model)
|
|
summary = await generate_completion(prompt_messages, model)
|
|
|
|
title = conv.title or "Conversation Summary"
|
|
title = f"Summary: {title}"
|
|
|
|
note = await create_note(user_id, title=title, body=summary, tags=["chat"])
|
|
return note.to_dict()
|