bvandeusen/FabledScribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 50 Wiki Activity

Add multi-user auth, background generation, and chat UX improvements

Browse Source 
Phase 5: Multi-user authentication with session cookies, bcrypt passwords,
first-user-is-admin pattern, per-user data isolation, backup/restore,
Docker Swarm production stack with secrets and network isolation.

Phase 5.1: Chat UX improvements:
- Background generation architecture (GenerationBuffer + asyncio task)
  with SSE fan-out, reconnect support, and periodic DB flushes
- LLM-generated conversation titles (first exchange + every 10th message)
- Stop generation button with cancel_event and partial content preservation
- Relative timestamps in sidebar (5m ago, 3h ago, then dates)
- Empty chat auto-cleanup on navigation away
- Save-as-note uses LLM for title generation, tags notes with "chat"
- Summarize-as-note also tags with "chat"

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Van Deusen
2026-02-11 14:36:30 -05:00
parent db01106714
commit cbfdf5289e
49 changed files with 3105 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fabledassistant/app.py
+33 -4
View File
@@ -1,10 +1,13 @@
import logging
import time
from pathlib import Path
from quart import Quart, jsonify, make_response, request, send_from_directory
from quart import Quart, g, jsonify, make_response, request, send_from_directory
from fabledassistant.config import Config
from fabledassistant.routes.admin import admin_bp
from fabledassistant.routes.api import api
from fabledassistant.routes.auth import auth_bp
from fabledassistant.routes.chat import chat_bp
from fabledassistant.routes.notes import notes_bp
from fabledassistant.routes.settings import settings_bp
@@ -15,21 +18,49 @@ logger = logging.getLogger(__name__)
def create_app() -> Quart:
# Configure logging
logging.basicConfig(
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
level=getattr(logging, Config.LOG_LEVEL.upper(), logging.INFO),
)
app = Quart(__name__, static_folder=None)
app.secret_key = Config.SECRET_KEY
app.register_blueprint(admin_bp)
app.register_blueprint(api)
app.register_blueprint(auth_bp)
app.register_blueprint(chat_bp)
app.register_blueprint(notes_bp)
app.register_blueprint(settings_bp)
app.register_blueprint(tasks_bp)
@app.before_request
async def before_request():
g.request_start = time.monotonic()
@app.after_request
async def after_request(response):
duration = time.monotonic() - getattr(g, "request_start", time.monotonic())
duration_ms = round(duration * 1000, 1)
logger.info(
"%s %s %s %.1fms",
request.method,
request.path,
response.status_code,
duration_ms,
)
return response
@app.before_serving
async def startup():
import asyncio
from fabledassistant.services.generation_buffer import start_cleanup_loop
from fabledassistant.services.llm import ensure_model
start_cleanup_loop()
async def _pull_model():
try:
await ensure_model(Config.OLLAMA_MODEL)
@@ -70,11 +101,9 @@ def create_app() -> Quart:
@app.errorhandler(500)
async def handle_500(error):
import traceback
traceback.print_exc()
logger.exception("Internal server error on %s %s", request.method, request.path)
if request.path.startswith("/api/"):
return jsonify({"error": str(error)}), 500
return jsonify({"error": "Internal server error"}), 500
return "Internal Server Error", 500
return app
src/fabledassistant/auth.py
+41
View File
@@ -0,0 +1,41 @@
import functools
from quart import g, jsonify, session
from fabledassistant.services.auth import get_user_by_id
def login_required(f):
@functools.wraps(f)
async def decorated(*args, **kwargs):
user_id = session.get("user_id")
if not user_id:
return jsonify({"error": "Authentication required"}), 401
user = await get_user_by_id(user_id)
if not user:
session.clear()
return jsonify({"error": "Authentication required"}), 401
g.user = user
return await f(*args, **kwargs)
return decorated
def admin_required(f):
@functools.wraps(f)
async def decorated(*args, **kwargs):
user_id = session.get("user_id")
if not user_id:
return jsonify({"error": "Authentication required"}), 401
user = await get_user_by_id(user_id)
if not user:
session.clear()
return jsonify({"error": "Authentication required"}), 401
if user.role != "admin":
return jsonify({"error": "Admin access required"}), 403
g.user = user
return await f(*args, **kwargs)
return decorated
def get_current_user_id() -> int:
return g.user.id
src/fabledassistant/config.py
+19 -2
View File
@@ -1,11 +1,28 @@
import os
def _read_secret(env_var: str, secret_file_var: str, default: str) -> str:
"""Read a config value from env var, or from a Docker secrets file."""
value = os.environ.get(env_var)
if value:
return value
secret_file = os.environ.get(secret_file_var)
if secret_file:
try:
with open(secret_file) as f:
return f.read().strip()
except FileNotFoundError:
pass
return default
class Config:
DATABASE_URL: str = os.environ.get(
DATABASE_URL: str = _read_secret(
"DATABASE_URL",
"DATABASE_URL_FILE",
"postgresql+asyncpg://fabled:fabled@localhost:5432/fabledassistant",
)
OLLAMA_URL: str = os.environ.get("OLLAMA_URL", "http://localhost:11434")
OLLAMA_MODEL: str = os.environ.get("OLLAMA_MODEL", "llama3.1")
SECRET_KEY: str = os.environ.get("SECRET_KEY", "dev-secret-change-me")
SECRET_KEY: str = _read_secret("SECRET_KEY", "SECRET_KEY_FILE", "dev-secret-change-me")
LOG_LEVEL: str = os.environ.get("LOG_LEVEL", "INFO")
src/fabledassistant/models/__init__.py
+1
View File
@@ -14,3 +14,4 @@ class Base(DeclarativeBase):
from fabledassistant.models.note import Note, TaskPriority, TaskStatus # noqa: E402, F401
from fabledassistant.models.conversation import Conversation, Message # noqa: E402, F401
from fabledassistant.models.setting import Setting # noqa: E402, F401
from fabledassistant.models.user import User # noqa: E402, F401
src/fabledassistant/models/conversation.py
+6
View File
@@ -11,6 +11,9 @@ class Conversation(Base):
__tablename__ = "conversations"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=True
)
title: Mapped[str] = mapped_column(Text, default="")
model: Mapped[str] = mapped_column(Text, default="")
created_at: Mapped[datetime] = mapped_column(
@@ -30,6 +33,7 @@ class Conversation(Base):
__table_args__ = (
Index("ix_conversations_updated_at", "updated_at"),
Index("ix_conversations_user_id", "user_id"),
)
def to_dict(self) -> dict:
@@ -57,6 +61,7 @@ class Message(Base):
)
role: Mapped[str] = mapped_column(Text)
content: Mapped[str] = mapped_column(Text, default="")
status: Mapped[str] = mapped_column(Text, default="complete")
context_note_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
)
@@ -76,6 +81,7 @@ class Message(Base):
"conversation_id": self.conversation_id,
"role": self.role,
"content": self.content,
"status": self.status,
"context_note_id": self.context_note_id,
"created_at": self.created_at.isoformat(),
}
src/fabledassistant/models/note.py
+5 -1
View File
@@ -1,7 +1,7 @@
import enum
from datetime import date, datetime, timezone
from sqlalchemy import Date, DateTime, Index, Text
from sqlalchemy import Date, DateTime, ForeignKey, Index, Integer, Text
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.orm import Mapped, mapped_column
@@ -25,6 +25,9 @@ class Note(Base):
__tablename__ = "notes"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=True
)
title: Mapped[str] = mapped_column(Text, default="")
body: Mapped[str] = mapped_column(Text, default="")
tags: Mapped[list[str]] = mapped_column(ARRAY(Text), default=list)
@@ -47,6 +50,7 @@ class Note(Base):
Index("ix_notes_tags", "tags", postgresql_using="gin"),
Index("ix_notes_status", "status"),
Index("ix_notes_title", "title"),
Index("ix_notes_user_id", "user_id"),
)
@property
src/fabledassistant/models/setting.py
+4 -1
View File
@@ -1,4 +1,4 @@
from sqlalchemy import Text
from sqlalchemy import ForeignKey, Integer, Text
from sqlalchemy.orm import Mapped, mapped_column
from fabledassistant.models import Base
@@ -7,6 +7,9 @@ from fabledassistant.models import Base
class Setting(Base):
__tablename__ = "settings"
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
key: Mapped[str] = mapped_column(Text, primary_key=True)
value: Mapped[str] = mapped_column(Text, default="")
src/fabledassistant/models/user.py
+32
View File
@@ -0,0 +1,32 @@
from datetime import datetime, timezone
from sqlalchemy import DateTime, Index, Text
from sqlalchemy.orm import Mapped, mapped_column
from fabledassistant.models import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(Text, unique=True, nullable=False)
email: Mapped[str | None] = mapped_column(Text, nullable=True)
password_hash: Mapped[str] = mapped_column(Text, nullable=False)
role: Mapped[str] = mapped_column(Text, nullable=False, default="user")
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
__table_args__ = (
Index("ix_users_username", "username"),
)
def to_dict(self) -> dict:
return {
"id": self.id,
"username": self.username,
"email": self.email,
"role": self.role,
"created_at": self.created_at.isoformat(),
}
src/fabledassistant/routes/admin.py
+47
View File
@@ -0,0 +1,47 @@
import json
from quart import Blueprint, Response, jsonify, request
from fabledassistant.auth import admin_required, login_required, get_current_user_id
from fabledassistant.services.backup import (
export_full_backup,
export_user_backup,
restore_full_backup,
)
admin_bp = Blueprint("admin", __name__, url_prefix="/api/admin")
@admin_bp.route("/backup", methods=["GET"])
@login_required
async def backup():
uid = get_current_user_id()
scope = request.args.get("scope", "user")
if scope == "full":
# Full backup requires admin
from quart import g
if g.user.role != "admin":
return jsonify({"error": "Admin access required for full backup"}), 403
data = await export_full_backup()
else:
data = await export_user_backup(uid)
return Response(
json.dumps(data, indent=2, default=str),
content_type="application/json",
headers={
"Content-Disposition": f'attachment; filename="fabled-backup-{scope}.json"',
},
)
@admin_bp.route("/restore", methods=["POST"])
@admin_required
async def restore():
data = await request.get_json()
if not data:
return jsonify({"error": "No backup data provided"}), 400
stats = await restore_full_backup(data)
return jsonify({"status": "ok", "stats": stats})
src/fabledassistant/routes/auth.py
+70
View File
@@ -0,0 +1,70 @@
from quart import Blueprint, jsonify, request, session
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.services.auth import (
authenticate,
create_user,
get_user_by_id,
get_user_count,
)
auth_bp = Blueprint("auth", __name__, url_prefix="/api/auth")
@auth_bp.route("/register", methods=["POST"])
async def register():
data = await request.get_json()
username = (data.get("username") or "").strip()
password = data.get("password") or ""
email = (data.get("email") or "").strip() or None
if not username:
return jsonify({"error": "Username is required"}), 400
if len(password) < 8:
return jsonify({"error": "Password must be at least 8 characters"}), 400
try:
user = await create_user(username, password, email)
except Exception:
return jsonify({"error": "Username already taken"}), 409
session["user_id"] = user.id
return jsonify(user.to_dict()), 201
@auth_bp.route("/login", methods=["POST"])
async def login():
data = await request.get_json()
username = (data.get("username") or "").strip()
password = data.get("password") or ""
if not username or not password:
return jsonify({"error": "Username and password are required"}), 400
user = await authenticate(username, password)
if not user:
return jsonify({"error": "Invalid username or password"}), 401
session["user_id"] = user.id
return jsonify(user.to_dict())
@auth_bp.route("/logout", methods=["POST"])
async def logout():
session.clear()
return jsonify({"status": "ok"})
@auth_bp.route("/me", methods=["GET"])
@login_required
async def me():
user = await get_user_by_id(get_current_user_id())
if not user:
return jsonify({"error": "User not found"}), 404
return jsonify(user.to_dict())
@auth_bp.route("/status", methods=["GET"])
async def status():
count = await get_user_count()
return jsonify({"has_users": count > 0})
src/fabledassistant/routes/chat.py
+115 -42
View File
@@ -1,9 +1,11 @@
import asyncio
import json
import logging
import httpx
from quart import Blueprint, Response, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.config import Config
from fabledassistant.services.chat import (
add_message,
@@ -15,7 +17,13 @@ from fabledassistant.services.chat import (
summarize_conversation_as_note,
update_conversation_title,
)
from fabledassistant.services.llm import build_context, stream_chat
from fabledassistant.services.generation_buffer import (
GenerationState,
create_buffer,
get_buffer,
)
from fabledassistant.services.generation_task import run_generation
from fabledassistant.services.llm import build_context
from fabledassistant.services.settings import get_setting
logger = logging.getLogger(__name__)
@@ -24,10 +32,12 @@ chat_bp = Blueprint("chat", __name__, url_prefix="/api/chat")
@chat_bp.route("/conversations", methods=["GET"])
@login_required
async def list_conversations_route():
uid = get_current_user_id()
limit = request.args.get("limit", 50, type=int)
offset = request.args.get("offset", 0, type=int)
conversations, total = await list_conversations(limit=limit, offset=offset)
conversations, total = await list_conversations(uid, limit=limit, offset=offset)
return jsonify({
"conversations": conversations,
"total": total,
@@ -35,17 +45,21 @@ async def list_conversations_route():
@chat_bp.route("/conversations", methods=["POST"])
@login_required
async def create_conversation_route():
uid = get_current_user_id()
data = await request.get_json(force=True, silent=True) or {}
title = data.get("title", "")
model = data.get("model", Config.OLLAMA_MODEL)
conv = await create_conversation(title=title, model=model)
conv = await create_conversation(uid, title=title, model=model)
return jsonify(conv.to_dict()), 201
@chat_bp.route("/conversations/<int:conv_id>", methods=["GET"])
@login_required
async def get_conversation_route(conv_id: int):
conv = await get_conversation(conv_id)
uid = get_current_user_id()
conv = await get_conversation(uid, conv_id)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
result = conv.to_dict()
@@ -54,29 +68,35 @@ async def get_conversation_route(conv_id: int):
@chat_bp.route("/conversations/<int:conv_id>", methods=["DELETE"])
@login_required
async def delete_conversation_route(conv_id: int):
deleted = await delete_conversation(conv_id)
uid = get_current_user_id()
deleted = await delete_conversation(uid, conv_id)
if not deleted:
return jsonify({"error": "Conversation not found"}), 404
return "", 204
@chat_bp.route("/conversations/<int:conv_id>", methods=["PATCH"])
@login_required
async def update_conversation_route(conv_id: int):
uid = get_current_user_id()
data = await request.get_json()
title = data.get("title")
if title is None:
return jsonify({"error": "title is required"}), 400
conv = await update_conversation_title(conv_id, title)
conv = await update_conversation_title(uid, conv_id, title)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
return jsonify(conv.to_dict())
@chat_bp.route("/conversations/<int:conv_id>/messages", methods=["POST"])
@login_required
async def send_message_route(conv_id: int):
"""Send a user message, stream the LLM response via SSE."""
conv = await get_conversation(conv_id)
"""Start generation: save user message, launch background task, return 202."""
uid = get_current_user_id()
conv = await get_conversation(uid, conv_id)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
@@ -87,10 +107,20 @@ async def send_message_route(conv_id: int):
context_note_id = data.get("context_note_id")
exclude_note_ids = data.get("exclude_note_ids") or []
# Reject if generation already running for this conversation
existing = get_buffer(conv_id)
if existing and existing.state == GenerationState.RUNNING:
return jsonify({"error": "Generation already in progress"}), 409
# Save user message
await add_message(conv_id, "user", content, context_note_id=context_note_id)
# Build history from existing messages (excluding system)
# Create placeholder assistant message
assistant_msg = await add_message(conv_id, "assistant", "", status="generating")
buf = create_buffer(conv_id, assistant_msg.id)
# Build history from existing messages (excluding system and the placeholder)
history = []
for msg in conv.messages:
if msg.role != "system":
@@ -98,43 +128,60 @@ async def send_message_route(conv_id: int):
# Build context with note search, URL fetching, etc.
messages, context_meta = await build_context(
history, context_note_id, content, exclude_note_ids=exclude_note_ids
uid, history, context_note_id, content, exclude_note_ids=exclude_note_ids
)
model = conv.model or await get_setting("default_model", Config.OLLAMA_MODEL)
model = conv.model or await get_setting(uid, "default_model", Config.OLLAMA_MODEL)
async def generate():
# Emit context metadata before streaming LLM response
context_event = json.dumps({"context": context_meta})
yield f"data: {context_event}\n\n"
# Launch background generation task
asyncio.create_task(run_generation(
buf, messages, model, context_meta,
uid, conv_id, conv.title, content,
))
full_response = []
try:
async for chunk in stream_chat(messages, model):
full_response.append(chunk)
event_data = json.dumps({"chunk": chunk})
yield f"data: {event_data}\n\n"
return jsonify({
"assistant_message_id": assistant_msg.id,
"status": "generating",
}), 202
# Save complete assistant message
full_text = "".join(full_response)
msg = await add_message(conv_id, "assistant", full_text)
# Auto-generate title from first user message if conversation has no title
if not conv.title:
title = content[:80]
if len(content) > 80:
title += "..."
await update_conversation_title(conv_id, title)
@chat_bp.route("/conversations/<int:conv_id>/generation/stream", methods=["GET"])
@login_required
async def generation_stream_route(conv_id: int):
"""SSE endpoint that tails the generation buffer for a conversation."""
uid = get_current_user_id()
conv = await get_conversation(uid, conv_id)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
done_data = json.dumps({"done": True, "message_id": msg.id})
yield f"data: {done_data}\n\n"
except Exception as e:
logger.exception("Error streaming LLM response")
error_data = json.dumps({"error": str(e)})
yield f"data: {error_data}\n\n"
buf = get_buffer(conv_id)
if buf is None:
return jsonify({"error": "No active generation"}), 404
# Determine starting point from Last-Event-ID header or query param
last_id_str = request.headers.get("Last-Event-ID") or request.args.get("last_event_id")
last_id = int(last_id_str) if last_id_str is not None else -1
async def stream():
cursor = last_id
while True:
# Replay any buffered events past the cursor
pending = buf.events_after(cursor)
for event in pending:
yield buf.format_sse(event)
cursor = event.index
# If generation is done and all events delivered, close stream
if buf.state != GenerationState.RUNNING:
break
# Wait for new events or send keepalive on timeout
got_new = await buf.wait_for_event(cursor, timeout=15.0)
if not got_new:
yield ": keepalive\n\n"
return Response(
generate(),
stream(),
content_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
@@ -143,32 +190,55 @@ async def send_message_route(conv_id: int):
)
@chat_bp.route("/conversations/<int:conv_id>/generation/cancel", methods=["POST"])
@login_required
async def cancel_generation_route(conv_id: int):
"""Cancel an active generation for a conversation."""
uid = get_current_user_id()
conv = await get_conversation(uid, conv_id)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
buf = get_buffer(conv_id)
if buf is None or buf.state != GenerationState.RUNNING:
return jsonify({"error": "No active generation"}), 404
buf.cancel_event.set()
return jsonify({"status": "cancelled"})
@chat_bp.route("/messages/<int:message_id>/save-as-note", methods=["POST"])
@login_required
async def save_message_as_note_route(message_id: int):
uid = get_current_user_id()
try:
note = await save_response_as_note(message_id)
note = await save_response_as_note(uid, message_id)
return jsonify(note), 201
except ValueError as e:
return jsonify({"error": str(e)}), 400
@chat_bp.route("/conversations/<int:conv_id>/summarize", methods=["POST"])
@login_required
async def summarize_conversation_route(conv_id: int):
conv = await get_conversation(conv_id)
uid = get_current_user_id()
conv = await get_conversation(uid, conv_id)
if conv is None:
return jsonify({"error": "Conversation not found"}), 404
model = conv.model or await get_setting("default_model", Config.OLLAMA_MODEL)
model = conv.model or await get_setting(uid, "default_model", Config.OLLAMA_MODEL)
try:
note = await summarize_conversation_as_note(conv_id, model)
note = await summarize_conversation_as_note(uid, conv_id, model)
return jsonify(note), 201
except ValueError as e:
return jsonify({"error": str(e)}), 400
@chat_bp.route("/status", methods=["GET"])
@login_required
async def chat_status_route():
"""Check Ollama availability and model readiness."""
default_model = await get_setting("default_model", Config.OLLAMA_MODEL)
uid = get_current_user_id()
default_model = await get_setting(uid, "default_model", Config.OLLAMA_MODEL)
result = {
"ollama": "unavailable",
"model": "not_found",
@@ -190,6 +260,7 @@ async def chat_status_route():
@chat_bp.route("/models", methods=["GET"])
@login_required
async def list_models_route():
try:
async with httpx.AsyncClient(timeout=10.0) as client:
@@ -207,6 +278,7 @@ async def list_models_route():
@chat_bp.route("/models/pull", methods=["POST"])
@login_required
async def pull_model_route():
"""Pull a model from Ollama, streaming progress via SSE."""
data = await request.get_json()
@@ -245,6 +317,7 @@ async def pull_model_route():
@chat_bp.route("/models/delete", methods=["POST"])
@login_required
async def delete_model_route():
"""Delete a model from Ollama."""
data = await request.get_json()
src/fabledassistant/routes/notes.py
+34 -10
View File
@@ -2,6 +2,7 @@ from datetime import date
from quart import Blueprint, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.services.notes import (
convert_note_to_task,
convert_task_to_note,
@@ -21,7 +22,9 @@ notes_bp = Blueprint("notes", __name__, url_prefix="/api/notes")
@notes_bp.route("", methods=["GET"])
@login_required
async def list_notes_route():
uid = get_current_user_id()
q = request.args.get("q")
tag = request.args.getlist("tag")
sort = request.args.get("sort", "updated_at")
@@ -37,13 +40,15 @@ async def list_notes_route():
is_task = True
notes, total = await list_notes(
q=q, tags=tag or None, is_task=is_task, sort=sort, order=order, limit=limit, offset=offset
uid, q=q, tags=tag or None, is_task=is_task, sort=sort, order=order, limit=limit, offset=offset
)
return jsonify({"notes": [n.to_dict() for n in notes], "total": total})
@notes_bp.route("", methods=["POST"])
@login_required
async def create_note_route():
uid = get_current_user_id()
data = await request.get_json()
body = data.get("body", "")
tags = extract_tags(body)
@@ -56,6 +61,7 @@ async def create_note_route():
due_date = date.fromisoformat(data["due_date"])
note = await create_note(
uid,
title=data.get("title", ""),
body=body,
tags=tags,
@@ -68,43 +74,53 @@ async def create_note_route():
@notes_bp.route("/tags", methods=["GET"])
@login_required
async def list_tags_route():
uid = get_current_user_id()
q = request.args.get("q")
tags = await get_all_tags(q=q)
tags = await get_all_tags(uid, q=q)
return jsonify({"tags": tags})
@notes_bp.route("/by-title", methods=["GET"])
@login_required
async def get_note_by_title_route():
uid = get_current_user_id()
title = request.args.get("title", "")
if not title:
return jsonify({"error": "title parameter is required"}), 400
note = await get_note_by_title(title)
note = await get_note_by_title(uid, title)
if note is None:
return jsonify({"error": "Note not found"}), 404
return jsonify(note.to_dict())
@notes_bp.route("/resolve-title", methods=["POST"])
@login_required
async def resolve_title_route():
uid = get_current_user_id()
data = await request.get_json()
title = data.get("title", "").strip()
if not title:
return jsonify({"error": "title is required"}), 400
note = await get_or_create_note_by_title(title)
note = await get_or_create_note_by_title(uid, title)
return jsonify(note.to_dict())
@notes_bp.route("/<int:note_id>", methods=["GET"])
@login_required
async def get_note_route(note_id: int):
note = await get_note(note_id)
uid = get_current_user_id()
note = await get_note(uid, note_id)
if note is None:
return jsonify({"error": "Note not found"}), 404
return jsonify(note.to_dict())
@notes_bp.route("/<int:note_id>", methods=["PUT"])
@login_required
async def update_note_route(note_id: int):
uid = get_current_user_id()
data = await request.get_json()
fields = {}
for key in ("title", "body", "parent_id", "status", "priority"):
@@ -118,39 +134,47 @@ async def update_note_route(note_id: int):
if "body" in fields:
fields["tags"] = extract_tags(fields["body"])
note = await update_note(note_id, **fields)
note = await update_note(uid, note_id, **fields)
if note is None:
return jsonify({"error": "Note not found"}), 404
return jsonify(note.to_dict())
@notes_bp.route("/<int:note_id>", methods=["DELETE"])
@login_required
async def delete_note_route(note_id: int):
deleted = await delete_note(note_id)
uid = get_current_user_id()
deleted = await delete_note(uid, note_id)
if not deleted:
return jsonify({"error": "Note not found"}), 404
return "", 204
@notes_bp.route("/<int:note_id>/convert-to-task", methods=["POST"])
@login_required
async def convert_note_to_task_route(note_id: int):
uid = get_current_user_id()
try:
note = await convert_note_to_task(note_id)
note = await convert_note_to_task(uid, note_id)
return jsonify(note.to_dict()), 200
except ValueError as e:
return jsonify({"error": str(e)}), 404
@notes_bp.route("/<int:note_id>/convert-to-note", methods=["POST"])
@login_required
async def convert_task_to_note_route(note_id: int):
uid = get_current_user_id()
try:
note = await convert_task_to_note(note_id)
note = await convert_task_to_note(uid, note_id)
return jsonify(note.to_dict()), 200
except ValueError as e:
return jsonify({"error": str(e)}), 404
@notes_bp.route("/<int:note_id>/backlinks", methods=["GET"])
@login_required
async def get_backlinks_route(note_id: int):
links = await get_backlinks(note_id)
uid = get_current_user_id()
links = await get_backlinks(uid, note_id)
return jsonify({"backlinks": links})
src/fabledassistant/routes/settings.py
+8 -3
View File
@@ -1,5 +1,6 @@
from quart import Blueprint, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.services.llm import get_installed_models
from fabledassistant.services.settings import get_all_settings, set_settings_batch
@@ -7,13 +8,17 @@ settings_bp = Blueprint("settings", __name__, url_prefix="/api/settings")
@settings_bp.route("", methods=["GET"])
@login_required
async def get_settings_route():
settings = await get_all_settings()
uid = get_current_user_id()
settings = await get_all_settings(uid)
return jsonify(settings)
@settings_bp.route("", methods=["PUT"])
@login_required
async def update_settings_route():
uid = get_current_user_id()
data = await request.get_json()
if not isinstance(data, dict):
return jsonify({"error": "Expected a JSON object"}), 400
@@ -24,6 +29,6 @@ async def update_settings_route():
if installed and model not in installed:
return jsonify({"error": f"Model '{model}' is not installed"}), 400
await set_settings_batch({k: str(v) for k, v in data.items()})
settings = await get_all_settings()
await set_settings_batch(uid, {k: str(v) for k, v in data.items()})
settings = await get_all_settings(uid)
return jsonify(settings)
src/fabledassistant/routes/tasks.py
+19 -4
View File
@@ -2,6 +2,7 @@ from datetime import date
from quart import Blueprint, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.models.note import TaskPriority, TaskStatus
from fabledassistant.services.notes import (
create_note,
@@ -16,7 +17,9 @@ tasks_bp = Blueprint("tasks", __name__, url_prefix="/api/tasks")
@tasks_bp.route("", methods=["GET"])
@login_required
async def list_tasks_route():
uid = get_current_user_id()
q = request.args.get("q")
tag = request.args.getlist("tag")
status = request.args.get("status")
@@ -27,6 +30,7 @@ async def list_tasks_route():
offset = request.args.get("offset", 0, type=int)
tasks, total = await list_notes(
uid,
q=q,
tags=tag or None,
is_task=True,
@@ -41,7 +45,9 @@ async def list_tasks_route():
@tasks_bp.route("", methods=["POST"])
@login_required
async def create_task_route():
uid = get_current_user_id()
data = await request.get_json()
body = data.get("body", "") or data.get("description", "")
tags = extract_tags(body)
@@ -56,6 +62,7 @@ async def create_task_route():
)
task = await create_note(
uid,
title=data.get("title", ""),
body=body,
status=status,
@@ -67,15 +74,19 @@ async def create_task_route():
@tasks_bp.route("/<int:task_id>", methods=["GET"])
@login_required
async def get_task_route(task_id: int):
task = await get_note(task_id)
uid = get_current_user_id()
task = await get_note(uid, task_id)
if task is None:
return jsonify({"error": "Task not found"}), 404
return jsonify(task.to_dict())
@tasks_bp.route("/<int:task_id>", methods=["PUT"])
@login_required
async def update_task_route(task_id: int):
uid = get_current_user_id()
data = await request.get_json()
fields = {}
for key in ("title", "status", "priority"):
@@ -96,14 +107,16 @@ async def update_task_route(task_id: int):
if "body" in fields:
fields["tags"] = extract_tags(fields["body"])
task = await update_note(task_id, **fields)
task = await update_note(uid, task_id, **fields)
if task is None:
return jsonify({"error": "Task not found"}), 404
return jsonify(task.to_dict())
@tasks_bp.route("/<int:task_id>/status", methods=["PATCH"])
@login_required
async def patch_task_status(task_id: int):
uid = get_current_user_id()
data = await request.get_json()
status_val = data.get("status")
if not status_val:
@@ -114,15 +127,17 @@ async def patch_task_status(task_id: int):
except ValueError:
return jsonify({"error": f"Invalid status: {status_val}"}), 400
task = await update_note(task_id, status=status_val)
task = await update_note(uid, task_id, status=status_val)
if task is None:
return jsonify({"error": "Task not found"}), 404
return jsonify(task.to_dict())
@tasks_bp.route("/<int:task_id>", methods=["DELETE"])
@login_required
async def delete_task_route(task_id: int):
deleted = await delete_note(task_id)
uid = get_current_user_id()
deleted = await delete_note(uid, task_id)
if not deleted:
return jsonify({"error": "Task not found"}), 404
return "", 204
src/fabledassistant/services/auth.py
+85
View File
@@ -0,0 +1,85 @@
import logging
import bcrypt
from sqlalchemy import func, select, update
from fabledassistant.models import async_session
from fabledassistant.models.conversation import Conversation
from fabledassistant.models.note import Note
from fabledassistant.models.setting import Setting
from fabledassistant.models.user import User
logger = logging.getLogger(__name__)
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
def verify_password(password: str, password_hash: str) -> bool:
return bcrypt.checkpw(password.encode(), password_hash.encode())
async def get_user_count() -> int:
async with async_session() as session:
return await session.scalar(select(func.count(User.id))) or 0
async def create_user(
username: str, password: str, email: str | None = None
) -> User:
user_count = await get_user_count()
role = "admin" if user_count == 0 else "user"
async with async_session() as session:
user = User(
username=username,
email=email,
password_hash=hash_password(password),
role=role,
)
session.add(user)
await session.commit()
await session.refresh(user)
# First user claims all pre-existing data (migration assigns user_id=1,
# which matches SERIAL first insert; also handles any NULLs)
if role == "admin":
await session.execute(
update(Note)
.where((Note.user_id.is_(None)) | (Note.user_id == user.id))
.values(user_id=user.id)
)
await session.execute(
update(Conversation)
.where(
(Conversation.user_id.is_(None))
| (Conversation.user_id == user.id)
)
.values(user_id=user.id)
)
await session.execute(
update(Setting)
.where(Setting.user_id == user.id)
.values(user_id=user.id)
)
await session.commit()
logger.info("First user '%s' created as admin, claimed orphaned data", username)
return user
async def authenticate(username: str, password: str) -> User | None:
async with async_session() as session:
result = await session.execute(
select(User).where(User.username == username)
)
user = result.scalars().first()
if user and verify_password(password, user.password_hash):
return user
return None
async def get_user_by_id(user_id: int) -> User | None:
async with async_session() as session:
return await session.get(User, user_id)
src/fabledassistant/services/backup.py
+255
View File
@@ -0,0 +1,255 @@
import logging
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from fabledassistant.models import async_session
from fabledassistant.models.conversation import Conversation, Message
from fabledassistant.models.note import Note
from fabledassistant.models.setting import Setting
from fabledassistant.models.user import User
logger = logging.getLogger(__name__)
async def export_full_backup() -> dict:
"""Export all users, notes, conversations+messages, and settings as JSON."""
async with async_session() as session:
users = (await session.execute(select(User))).scalars().all()
notes = (await session.execute(select(Note))).scalars().all()
conversations = (
await session.execute(
select(Conversation).options(selectinload(Conversation.messages))
)
).scalars().all()
settings = (await session.execute(select(Setting))).scalars().all()
return {
"version": 1,
"scope": "full",
"users": [
{
"id": u.id,
"username": u.username,
"email": u.email,
"password_hash": u.password_hash,
"role": u.role,
"created_at": u.created_at.isoformat(),
}
for u in users
],
"notes": [
{
"id": n.id,
"user_id": n.user_id,
"title": n.title,
"body": n.body,
"tags": n.tags or [],
"parent_id": n.parent_id,
"status": n.status,
"priority": n.priority,
"due_date": n.due_date.isoformat() if n.due_date else None,
"created_at": n.created_at.isoformat(),
"updated_at": n.updated_at.isoformat(),
}
for n in notes
],
"conversations": [
{
"id": c.id,
"user_id": c.user_id,
"title": c.title,
"model": c.model,
"created_at": c.created_at.isoformat(),
"updated_at": c.updated_at.isoformat(),
"messages": [
{
"id": m.id,
"role": m.role,
"content": m.content,
"context_note_id": m.context_note_id,
"created_at": m.created_at.isoformat(),
}
for m in c.messages
],
}
for c in conversations
],
"settings": [
{
"user_id": s.user_id,
"key": s.key,
"value": s.value,
}
for s in settings
],
}
async def export_user_backup(user_id: int) -> dict:
"""Export a single user's data as JSON."""
async with async_session() as session:
user = await session.get(User, user_id)
notes = (
await session.execute(select(Note).where(Note.user_id == user_id))
).scalars().all()
conversations = (
await session.execute(
select(Conversation)
.options(selectinload(Conversation.messages))
.where(Conversation.user_id == user_id)
)
).scalars().all()
settings = (
await session.execute(
select(Setting).where(Setting.user_id == user_id)
)
).scalars().all()
return {
"version": 1,
"scope": "user",
"user": {
"id": user.id,
"username": user.username,
"email": user.email,
"role": user.role,
"created_at": user.created_at.isoformat(),
} if user else None,
"notes": [
{
"id": n.id,
"title": n.title,
"body": n.body,
"tags": n.tags or [],
"parent_id": n.parent_id,
"status": n.status,
"priority": n.priority,
"due_date": n.due_date.isoformat() if n.due_date else None,
"created_at": n.created_at.isoformat(),
"updated_at": n.updated_at.isoformat(),
}
for n in notes
],
"conversations": [
{
"id": c.id,
"title": c.title,
"model": c.model,
"created_at": c.created_at.isoformat(),
"updated_at": c.updated_at.isoformat(),
"messages": [
{
"id": m.id,
"role": m.role,
"content": m.content,
"context_note_id": m.context_note_id,
"created_at": m.created_at.isoformat(),
}
for m in c.messages
],
}
for c in conversations
],
"settings": [
{"key": s.key, "value": s.value}
for s in settings
],
}
async def restore_full_backup(data: dict) -> dict:
"""Restore from a full backup JSON. Returns stats about what was restored."""
from datetime import date, datetime, timezone
stats = {"users": 0, "notes": 0, "conversations": 0, "messages": 0, "settings": 0}
async with async_session() as session:
# Restore users
user_id_map: dict[int, int] = {}
for u_data in data.get("users", []):
old_id = u_data["id"]
user = User(
username=u_data["username"],
email=u_data.get("email"),
password_hash=u_data["password_hash"],
role=u_data.get("role", "user"),
created_at=datetime.fromisoformat(u_data["created_at"]) if u_data.get("created_at") else datetime.now(timezone.utc),
)
session.add(user)
await session.flush()
user_id_map[old_id] = user.id
stats["users"] += 1
# Restore notes
note_id_map: dict[int, int] = {}
for n_data in data.get("notes", []):
old_id = n_data.get("id")
mapped_user_id = user_id_map.get(n_data.get("user_id", 0))
if mapped_user_id is None:
continue
due = None
if n_data.get("due_date"):
due = date.fromisoformat(n_data["due_date"])
note = Note(
user_id=mapped_user_id,
title=n_data.get("title", ""),
body=n_data.get("body", ""),
tags=n_data.get("tags", []),
parent_id=n_data.get("parent_id"),
status=n_data.get("status"),
priority=n_data.get("priority"),
due_date=due,
created_at=datetime.fromisoformat(n_data["created_at"]) if n_data.get("created_at") else datetime.now(timezone.utc),
updated_at=datetime.fromisoformat(n_data["updated_at"]) if n_data.get("updated_at") else datetime.now(timezone.utc),
)
session.add(note)
await session.flush()
if old_id is not None:
note_id_map[old_id] = note.id
stats["notes"] += 1
# Restore conversations + messages
for c_data in data.get("conversations", []):
mapped_user_id = user_id_map.get(c_data.get("user_id", 0))
if mapped_user_id is None:
continue
conv = Conversation(
user_id=mapped_user_id,
title=c_data.get("title", ""),
model=c_data.get("model", ""),
created_at=datetime.fromisoformat(c_data["created_at"]) if c_data.get("created_at") else datetime.now(timezone.utc),
updated_at=datetime.fromisoformat(c_data["updated_at"]) if c_data.get("updated_at") else datetime.now(timezone.utc),
)
session.add(conv)
await session.flush()
stats["conversations"] += 1
for m_data in c_data.get("messages", []):
msg = Message(
conversation_id=conv.id,
role=m_data["role"],
content=m_data.get("content", ""),
context_note_id=note_id_map.get(m_data.get("context_note_id")) if m_data.get("context_note_id") else None,
created_at=datetime.fromisoformat(m_data["created_at"]) if m_data.get("created_at") else datetime.now(timezone.utc),
)
session.add(msg)
stats["messages"] += 1
# Restore settings
for s_data in data.get("settings", []):
mapped_user_id = user_id_map.get(s_data.get("user_id", 0))
if mapped_user_id is None:
continue
setting = Setting(
user_id=mapped_user_id,
key=s_data["key"],
value=s_data.get("value", ""),
)
session.add(setting)
stats["settings"] += 1
await session.commit()
logger.info("Restored backup: %s", stats)
return stats
src/fabledassistant/services/chat.py
+68 -20
View File
@@ -12,9 +12,11 @@ from fabledassistant.services.notes import create_note
logger = logging.getLogger(__name__)
async def create_conversation(title: str = "", model: str = "") -> Conversation:
async def create_conversation(
user_id: int, title: str = "", model: str = ""
) -> Conversation:
async with async_session() as session:
conv = Conversation(title=title, model=model)
conv = Conversation(user_id=user_id, title=title, model=model)
session.add(conv)
await session.commit()
# Re-fetch with messages eagerly loaded to avoid lazy-load in async context
@@ -26,22 +28,29 @@ async def create_conversation(title: str = "", model: str = "") -> Conversation:
return result.scalars().first()
async def get_conversation(conversation_id: int) -> Conversation | None:
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)
.where(
Conversation.id == conversation_id,
Conversation.user_id == user_id,
)
)
return result.scalars().first()
async def list_conversations(
limit: int = 50, offset: int = 0
user_id: int, limit: int = 50, offset: int = 0
) -> tuple[list[dict], int]:
async with async_session() as session:
total = await session.scalar(
select(func.count(Conversation.id))
select(func.count(Conversation.id)).where(
Conversation.user_id == user_id
)
) or 0
# Subquery for message count — avoids loading all messages
@@ -54,6 +63,7 @@ async def list_conversations(
result = await session.execute(
select(Conversation, msg_count.label("message_count"))
.where(Conversation.user_id == user_id)
.order_by(Conversation.updated_at.desc())
.limit(limit)
.offset(offset)
@@ -75,9 +85,15 @@ async def list_conversations(
return conversations, total
async def delete_conversation(conversation_id: int) -> bool:
async def delete_conversation(user_id: int, conversation_id: int) -> bool:
async with async_session() as session:
conv = await session.get(Conversation, conversation_id)
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)
@@ -86,10 +102,16 @@ async def delete_conversation(conversation_id: int) -> bool:
async def update_conversation_title(
conversation_id: int, title: str
user_id: int, conversation_id: int, title: str
) -> Conversation | None:
async with async_session() as session:
conv = await session.get(Conversation, conversation_id)
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
conv.title = title
@@ -104,14 +126,18 @@ async def add_message(
role: str,
content: str,
context_note_id: int | None = None,
status: str | None = None,
) -> Message:
async with async_session() as session:
msg = Message(
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
msg = Message(**kwargs)
session.add(msg)
# Touch conversation updated_at
conv = await session.get(Conversation, conversation_id)
@@ -127,7 +153,7 @@ async def get_message(message_id: int) -> Message | None:
return await session.get(Message, message_id)
async def save_response_as_note(message_id: int) -> dict:
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:
@@ -135,20 +161,42 @@ async def save_response_as_note(message_id: int) -> dict:
if msg.role != "assistant":
raise ValueError("Can only save assistant messages as notes")
# Use first line as title, rest as body
lines = msg.content.strip().split("\n", 1)
title = lines[0].strip().lstrip("# ")[:100]
body = msg.content
conv = await get_conversation(user_id, msg.conversation_id)
note = await create_note(title=title, body=body)
# Generate title via LLM using the assistant message content
title = ""
if conv:
model = conv.model or "llama3.2"
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]},
]
title = await generate_completion(prompt_messages, 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(
conversation_id: int, model: str
user_id: int, conversation_id: int, model: str
) -> dict:
"""Summarize a conversation using the LLM and save as a note."""
conv = await get_conversation(conversation_id)
conv = await get_conversation(user_id, conversation_id)
if conv is None:
raise ValueError("Conversation not found")
@@ -178,5 +226,5 @@ async def summarize_conversation_as_note(
title = conv.title or "Conversation Summary"
title = f"Summary: {title}"
note = await create_note(title=title, body=summary)
note = await create_note(user_id, title=title, body=summary, tags=["chat"])
return note.to_dict()
src/fabledassistant/services/generation_buffer.py
+114
View File
@@ -0,0 +1,114 @@
"""In-memory generation event buffer for SSE fan-out.
Each active generation gets a GenerationBuffer keyed by conversation_id.
SSE clients tail the buffer and can reconnect at any point without data loss.
"""
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
logger = logging.getLogger(__name__)
class GenerationState(str, Enum):
RUNNING = "running"
COMPLETED = "completed"
ERRORED = "errored"
@dataclass
class SSEEvent:
index: int
event_type: str # "context", "chunk", "done", "error"
data: dict
@dataclass
class GenerationBuffer:
conversation_id: int
assistant_message_id: int
state: GenerationState = GenerationState.RUNNING
events: list[SSEEvent] = field(default_factory=list)
content_so_far: str = ""
finished_at: float | None = None
cancel_event: asyncio.Event = field(default_factory=asyncio.Event)
_notify: asyncio.Event = field(default_factory=asyncio.Event)
def append_event(self, event_type: str, data: dict) -> SSEEvent:
event = SSEEvent(index=len(self.events), event_type=event_type, data=data)
self.events.append(event)
# Wake all waiting SSE clients, then reset for next wait
old = self._notify
self._notify = asyncio.Event()
old.set()
return event
async def wait_for_event(self, after_index: int, timeout: float = 30.0) -> bool:
"""Wait until there are events beyond after_index. Returns True if new events exist."""
if after_index + 1 < len(self.events):
return True
try:
await asyncio.wait_for(self._notify.wait(), timeout=timeout)
return True
except asyncio.TimeoutError:
return False
def events_after(self, index: int) -> list[SSEEvent]:
"""Return events with index > given index (for replay on reconnect)."""
start = index + 1
if start >= len(self.events):
return []
return self.events[start:]
@staticmethod
def format_sse(event: SSEEvent) -> str:
return f"id: {event.index}\nevent: {event.event_type}\ndata: {json.dumps(event.data)}\n\n"
# Module-level singleton registry
_buffers: dict[int, GenerationBuffer] = {}
_cleanup_task: asyncio.Task | None = None
_GRACE_PERIOD = 60.0 # seconds to keep completed buffers
def create_buffer(conv_id: int, msg_id: int) -> GenerationBuffer:
existing = _buffers.get(conv_id)
if existing and existing.state == GenerationState.RUNNING:
raise RuntimeError(f"Generation already running for conversation {conv_id}")
buf = GenerationBuffer(conversation_id=conv_id, assistant_message_id=msg_id)
_buffers[conv_id] = buf
return buf
def get_buffer(conv_id: int) -> GenerationBuffer | None:
return _buffers.get(conv_id)
def remove_buffer(conv_id: int) -> None:
_buffers.pop(conv_id, None)
async def _cleanup_loop() -> None:
"""Remove completed/errored buffers after grace period."""
while True:
await asyncio.sleep(15)
now = time.monotonic()
to_remove = [
cid for cid, buf in _buffers.items()
if buf.state != GenerationState.RUNNING
and buf.finished_at is not None
and now - buf.finished_at > _GRACE_PERIOD
]
for cid in to_remove:
_buffers.pop(cid, None)
logger.debug("Cleaned up generation buffer for conversation %d", cid)
def start_cleanup_loop() -> None:
global _cleanup_task
if _cleanup_task is None or _cleanup_task.done():
_cleanup_task = asyncio.create_task(_cleanup_loop())
src/fabledassistant/services/generation_task.py
+136
View File
@@ -0,0 +1,136 @@
"""Background asyncio task for LLM generation.
Streams from Ollama into a GenerationBuffer, periodically flushing to DB.
Runs independently of any HTTP connection.
"""
import logging
import time
from sqlalchemy import update
from fabledassistant.models import async_session
from fabledassistant.models.conversation import Message
from fabledassistant.services.generation_buffer import GenerationBuffer, GenerationState
from fabledassistant.services.llm import generate_completion, stream_chat
from fabledassistant.services.chat import update_conversation_title
logger = logging.getLogger(__name__)
DB_FLUSH_INTERVAL = 5.0 # seconds between partial DB flushes
async def _generate_title(messages: list[dict], model: str) -> str:
"""Ask the LLM for a concise conversation title."""
# Build conversation text like summarize_conversation_as_note
conv_lines = []
for m in messages:
if m["role"] == "system":
continue
label = "User" if m["role"] == "user" else "Assistant"
conv_lines.append(f"{label}: {m['content']}")
# Keep only last 6 pairs worth of text
conv_lines = conv_lines[-12:]
prompt_messages = [
{
"role": "system",
"content": (
"Generate a concise 3-8 word title for this conversation. "
"Reply with ONLY the title, no quotes or punctuation."
),
},
{"role": "user", "content": "\n\n".join(conv_lines)},
]
title = await generate_completion(prompt_messages, model)
title = title.strip().strip('"\'').strip()
return title[:100] if title else ""
async def _update_message(message_id: int, content: str, status: str) -> None:
async with async_session() as session:
await session.execute(
update(Message)
.where(Message.id == message_id)
.values(content=content, status=status)
)
await session.commit()
async def run_generation(
buf: GenerationBuffer,
messages: list[dict],
model: str,
context_meta: dict,
user_id: int,
conv_id: int,
conv_title: str,
user_content: str,
) -> None:
"""Stream LLM response into buffer with periodic DB flushes."""
msg_id = buf.assistant_message_id
# Emit context event
buf.append_event("context", {"context": context_meta})
last_flush = time.monotonic()
try:
cancelled = False
async for chunk in stream_chat(messages, model):
if buf.cancel_event.is_set():
cancelled = True
break
buf.content_so_far += chunk
buf.append_event("chunk", {"chunk": chunk})
# Periodic DB flush
now = time.monotonic()
if now - last_flush >= DB_FLUSH_INTERVAL:
try:
await _update_message(msg_id, buf.content_so_far, "generating")
except Exception:
logger.warning("Failed periodic flush for message %d", msg_id, exc_info=True)
last_flush = now
# Final save (partial content on cancel is still valid)
await _update_message(msg_id, buf.content_so_far, "complete")
# Count non-system messages to decide on title generation
non_system = [m for m in messages if m["role"] != "system"]
msg_count = len(non_system)
should_gen_title = not conv_title or (msg_count > 0 and msg_count % 10 == 0)
if should_gen_title:
# Include the just-generated assistant reply for context
title_messages = messages + [
{"role": "assistant", "content": buf.content_so_far}
]
try:
title = await _generate_title(title_messages, model)
if title:
await update_conversation_title(user_id, conv_id, title)
except Exception:
logger.warning("Failed to generate title for conversation %d", conv_id, exc_info=True)
# Fallback for first message only
if not conv_title:
fallback = user_content[:80]
if len(user_content) > 80:
fallback += "..."
await update_conversation_title(user_id, conv_id, fallback)
buf.state = GenerationState.COMPLETED
buf.finished_at = time.monotonic()
buf.append_event("done", {"done": True, "message_id": msg_id})
except Exception as e:
logger.exception("Error in generation task for conversation %d", conv_id)
# Save partial content with error status
try:
await _update_message(msg_id, buf.content_so_far, "error")
except Exception:
logger.warning("Failed to save error state for message %d", msg_id, exc_info=True)
buf.state = GenerationState.ERRORED
buf.finished_at = time.monotonic()
buf.append_event("error", {"error": str(e)})
src/fabledassistant/services/llm.py
+4 -3
View File
@@ -149,6 +149,7 @@ def _find_urls(text: str) -> list[str]:
async def build_context(
user_id: int,
history: list[dict],
current_note_id: int | None,
user_message: str,
@@ -160,7 +161,7 @@ async def build_context(
which notes were included as context.
"""
exclude_set = set(exclude_note_ids or [])
assistant_name = await get_setting("assistant_name", "Fable")
assistant_name = await get_setting(user_id, "assistant_name", "Fable")
system_parts = [
f"You are a helpful assistant named {assistant_name}, integrated into a note-taking and task-tracking app called Fabled Assistant. "
"Help users with their notes, tasks, and general questions. "
@@ -175,7 +176,7 @@ async def build_context(
# Include current note context if provided
if current_note_id:
note = await get_note(current_note_id)
note = await get_note(user_id, current_note_id)
if note:
context_meta["context_note_id"] = note.id
context_meta["context_note_title"] = note.title
@@ -194,7 +195,7 @@ async def build_context(
search_exclude.add(current_note_id)
try:
notes = await search_notes_for_context(
keywords, exclude_ids=search_exclude or None, limit=3
user_id, keywords, exclude_ids=search_exclude or None, limit=3
)
snippets: list[str] = []
for n in notes:
src/fabledassistant/services/notes.py
+48 -22
View File
@@ -10,6 +10,7 @@ logger = logging.getLogger(__name__)
async def create_note(
user_id: int,
title: str = "",
body: str = "",
tags: list[str] | None = None,
@@ -20,6 +21,7 @@ async def create_note(
) -> Note:
async with async_session() as session:
note = Note(
user_id=user_id,
title=title,
body=body,
tags=tags or [],
@@ -34,12 +36,16 @@ async def create_note(
return note
async def get_note(note_id: int) -> Note | None:
async def get_note(user_id: int, note_id: int) -> Note | None:
async with async_session() as session:
return await session.get(Note, note_id)
result = await session.execute(
select(Note).where(Note.id == note_id, Note.user_id == user_id)
)
return result.scalars().first()
async def list_notes(
user_id: int,
q: str | None = None,
tags: list[str] | None = None,
is_task: bool | None = None,
@@ -51,8 +57,8 @@ async def list_notes(
offset: int = 0,
) -> tuple[list[Note], int]:
async with async_session() as session:
query = select(Note)
count_query = select(func.count(Note.id))
query = select(Note).where(Note.user_id == user_id)
count_query = select(func.count(Note.id)).where(Note.user_id == user_id)
# Filter by task vs note
if is_task is True:
@@ -104,25 +110,31 @@ async def list_notes(
return notes, total
async def get_note_by_title(title: str) -> Note | None:
async def get_note_by_title(user_id: int, title: str) -> Note | None:
async with async_session() as session:
result = await session.execute(
select(Note).where(func.lower(Note.title) == func.lower(title.strip())).limit(1)
select(Note).where(
Note.user_id == user_id,
func.lower(Note.title) == func.lower(title.strip()),
).limit(1)
)
return result.scalars().first()
async def get_or_create_note_by_title(title: str) -> Note:
async def get_or_create_note_by_title(user_id: int, title: str) -> Note:
title = title.strip()
note = await get_note_by_title(title)
note = await get_note_by_title(user_id, title)
if note:
return note
return await create_note(title=title)
return await create_note(user_id, title=title)
async def update_note(note_id: int, **fields: object) -> Note | None:
async def update_note(user_id: int, note_id: int, **fields: object) -> Note | None:
async with async_session() as session:
note = await session.get(Note, note_id)
result = await session.execute(
select(Note).where(Note.id == note_id, Note.user_id == user_id)
)
note = result.scalars().first()
if note is None:
return None
for key, value in fields.items():
@@ -139,9 +151,12 @@ async def update_note(note_id: int, **fields: object) -> Note | None:
return note
async def delete_note(note_id: int) -> bool:
async def delete_note(user_id: int, note_id: int) -> bool:
async with async_session() as session:
note = await session.get(Note, note_id)
result = await session.execute(
select(Note).where(Note.id == note_id, Note.user_id == user_id)
)
note = result.scalars().first()
if note is None:
return False
await session.delete(note)
@@ -149,10 +164,13 @@ async def delete_note(note_id: int) -> bool:
return True
async def get_all_tags(q: str | None = None) -> list[str]:
async def get_all_tags(user_id: int, q: str | None = None) -> list[str]:
async with async_session() as session:
result = await session.execute(
text("SELECT DISTINCT unnest(tags) AS tag FROM notes WHERE tags != '{}'")
text(
"SELECT DISTINCT unnest(tags) AS tag FROM notes"
" WHERE tags != '{}' AND user_id = :user_id"
).bindparams(user_id=user_id)
)
all_tags = [row[0] for row in result.fetchall()]
@@ -163,9 +181,12 @@ async def get_all_tags(q: str | None = None) -> list[str]:
return sorted(all_tags)
async def convert_note_to_task(note_id: int) -> Note:
async def convert_note_to_task(user_id: int, note_id: int) -> Note:
async with async_session() as session:
note = await session.get(Note, note_id)
result = await session.execute(
select(Note).where(Note.id == note_id, Note.user_id == user_id)
)
note = result.scalars().first()
if note is None:
logger.warning("convert_note_to_task: note %d not found", note_id)
raise ValueError("Note not found")
@@ -178,9 +199,12 @@ async def convert_note_to_task(note_id: int) -> Note:
return note
async def convert_task_to_note(note_id: int) -> Note:
async def convert_task_to_note(user_id: int, note_id: int) -> Note:
async with async_session() as session:
note = await session.get(Note, note_id)
result = await session.execute(
select(Note).where(Note.id == note_id, Note.user_id == user_id)
)
note = result.scalars().first()
if note is None:
logger.warning("convert_task_to_note: note %d not found", note_id)
raise ValueError("Note not found")
@@ -195,6 +219,7 @@ async def convert_task_to_note(note_id: int) -> Note:
async def search_notes_for_context(
user_id: int,
keywords: list[str],
exclude_ids: set[int] | None = None,
limit: int = 3,
@@ -207,7 +232,7 @@ async def search_notes_for_context(
pattern = f"%{escaped}%"
keyword_filters.append(or_(Note.title.ilike(pattern), Note.body.ilike(pattern)))
query = select(Note).where(or_(*keyword_filters))
query = select(Note).where(Note.user_id == user_id, or_(*keyword_filters))
if exclude_ids:
query = query.where(Note.id.notin_(exclude_ids))
query = query.order_by(Note.updated_at.desc()).limit(limit)
@@ -216,8 +241,8 @@ async def search_notes_for_context(
return list(result.scalars().all())
async def get_backlinks(note_id: int) -> list[dict]:
note = await get_note(note_id)
async def get_backlinks(user_id: int, note_id: int) -> list[dict]:
note = await get_note(user_id, note_id)
if note is None:
return []
@@ -232,6 +257,7 @@ async def get_backlinks(note_id: int) -> list[dict]:
results = await session.execute(
select(Note.id, Note.title, Note.status).where(
Note.user_id == user_id,
Note.id != note_id,
or_(Note.body.like(pattern), Note.body.like(pattern_alias)),
)
src/fabledassistant/services/settings.py
+19 -11
View File
@@ -8,39 +8,47 @@ from fabledassistant.models.setting import Setting
logger = logging.getLogger(__name__)
async def get_setting(key: str, default: str = "") -> str:
async def get_setting(user_id: int, key: str, default: str = "") -> str:
async with async_session() as session:
result = await session.execute(select(Setting).where(Setting.key == key))
result = await session.execute(
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
)
setting = result.scalar_one_or_none()
return setting.value if setting else default
async def set_setting(key: str, value: str) -> None:
async def set_setting(user_id: int, key: str, value: str) -> None:
async with async_session() as session:
result = await session.execute(select(Setting).where(Setting.key == key))
result = await session.execute(
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
)
setting = result.scalar_one_or_none()
if setting:
setting.value = value
else:
session.add(Setting(key=key, value=value))
session.add(Setting(user_id=user_id, key=key, value=value))
await session.commit()
async def set_settings_batch(settings: dict[str, str]) -> None:
async def set_settings_batch(user_id: int, settings: dict[str, str]) -> None:
"""Update multiple settings in a single transaction."""
async with async_session() as session:
for key, value in settings.items():
result = await session.execute(select(Setting).where(Setting.key == key))
result = await session.execute(
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
)
setting = result.scalar_one_or_none()
if setting:
setting.value = value
else:
session.add(Setting(key=key, value=value))
session.add(Setting(user_id=user_id, key=key, value=value))
await session.commit()
logger.info("Batch-updated %d settings", len(settings))
logger.info("Batch-updated %d settings for user %d", len(settings), user_id)
async def get_all_settings() -> dict[str, str]:
async def get_all_settings(user_id: int) -> dict[str, str]:
async with async_session() as session:
result = await session.execute(select(Setting))
result = await session.execute(
select(Setting).where(Setting.user_id == user_id)
)
return {s.key: s.value for s in result.scalars().all()}