Background generation, unread replies, send/stop, PLAN.md
**Replies now run in the background.** Generation was driven by the SSE request, so navigating away or opening another chat cut the answer off mid-sentence. services/generation.py owns the work as its own task and the SSE endpoint merely follows it. Verified: attached briefly, closed the connection, went to another page -- the reply finished anyway, 832 characters, not marked stopped, auto-titled. Reattaching works because both `render` and `reasoning` frames now carry the whole block rather than a delta. A follower arriving late has no earlier fragments to append to, so deltas would leave it permanently missing the beginning. Verified: attached six seconds in and the first frame already contained 517 characters written while nobody watched. **Unread indicator.** A reply that lands with no follower attached marks its chat unread; the sidebar polls every 10s for out-of-band dot spans plus an HX-Trigger that raises a toast. Polled rather than pushed: a browser sitting on another chat has no connection to the one that finished, and an always-on channel per tab is a lot of machinery for a green dot. `unread_notified` stops the same arrival being announced every tick. Follower count is what decides "was anyone watching", so reading it as it arrives does not mark it unread -- verified both ways. **Stop is the send button.** While a reply is being written the send button becomes a red stop square, found via a MutationObserver on the thread since the composer and the streaming bubble are far apart in the document. The in-bubble Stop is gone. **Attachment border removed.** As asked -- an attachment is a picture, and the frame only ever drew at the wrong width. The anchor now shrink-wraps and the img's width/height attributes are overridden so a small image shows at its own size. Adds PLAN.md: what is built, what is not, known limits, and the decisions that look like oversights until you know the reason. 239 tests, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
+82
-178
@@ -3,8 +3,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
|
||||
@@ -18,37 +18,15 @@ from lembas.db.session import session_scope
|
||||
from lembas.security import permissions
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services import files as files_service
|
||||
from lembas.services import generation as generation_service
|
||||
from lembas.services import sse
|
||||
from lembas.services.llm.openai_client import (
|
||||
LLMError,
|
||||
delta_reasoning,
|
||||
delta_text,
|
||||
stream_chat,
|
||||
)
|
||||
from lembas.services.markdown import escape_text, render_markdown
|
||||
from lembas.services.reasoning import REASONING, ReasoningSplitter
|
||||
from lembas.web.templating import render, templates
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/chats", tags=["chats"])
|
||||
|
||||
# Message ids whose generation has been asked to stop. The generator checks
|
||||
# this between chunks and finalises with whatever it has.
|
||||
#
|
||||
# In-process, which is correct for the single-worker deployment this ships
|
||||
# with: the request that stops a stream and the task producing it are in the
|
||||
# same process. Running multiple workers would need this in the database or a
|
||||
# broker instead -- see deploy/README.md.
|
||||
_CANCELLED: set[str] = set()
|
||||
|
||||
# How often the partially rendered reply is pushed to the browser. Markdown is
|
||||
# re-rendered from scratch each time, so this trades a little server work for
|
||||
# formatting that appears as the model writes rather than all at once at the
|
||||
# end. 100ms is below the threshold where the eye reads it as stepping.
|
||||
RENDER_INTERVAL = 0.1
|
||||
|
||||
|
||||
def _owned_chat(db: DBSession, chat_id: str, user_id: str) -> Chat:
|
||||
chat = db.get(Chat, chat_id)
|
||||
# 404 rather than 403 for someone else's chat: whether a given id exists is
|
||||
@@ -106,9 +84,10 @@ async def start_chat(
|
||||
user_message = chat_service.create_message(db, chat, ROLE_USER, content)
|
||||
if file_ids:
|
||||
files_service.claim(db, ids=file_ids, user_id=user.id, message_id=user_message.id)
|
||||
chat_service.create_message(
|
||||
assistant = chat_service.create_message(
|
||||
db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=chat.model_id
|
||||
)
|
||||
generation_service.ensure(chat.id, assistant.id)
|
||||
|
||||
response = Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
response.headers["HX-Redirect"] = f"/chat/{chat.id}"
|
||||
@@ -120,6 +99,44 @@ async def start_chat(
|
||||
# /start when the first message is actually sent.
|
||||
|
||||
|
||||
@router.get("/unread")
|
||||
async def unread_poll(db: Db, user: RequiredUser) -> Response:
|
||||
"""Dots for the sidebar, and a toast for anything newly arrived.
|
||||
|
||||
Polled rather than pushed: a browser sitting on a different chat has no
|
||||
open connection to the one that finished, and a second always-on channel
|
||||
per tab is a lot of machinery for a green dot.
|
||||
|
||||
Returns out-of-band spans so only the dots change -- re-rendering the whole
|
||||
sidebar would reset the folder open/closed state on every tick.
|
||||
"""
|
||||
chats = list(
|
||||
db.scalars(
|
||||
select(Chat).where(Chat.user_id == user.id, Chat.archived.is_(False))
|
||||
)
|
||||
)
|
||||
|
||||
fresh = [c for c in chats if c.unread and not c.unread_notified]
|
||||
for chat in fresh:
|
||||
chat.unread_notified = True
|
||||
if fresh:
|
||||
db.commit()
|
||||
|
||||
markup = "".join(
|
||||
f'<span id="unread-{c.id}" class="unread-dot" hx-swap-oob="true"'
|
||||
f'{"" if c.unread else " hidden"} title="New reply"></span>'
|
||||
for c in chats
|
||||
)
|
||||
|
||||
response = HTMLResponse(markup)
|
||||
if fresh:
|
||||
# HX-Trigger carries the toast; ui.js listens for it.
|
||||
response.headers["HX-Trigger"] = json.dumps(
|
||||
{"lembas:unread": {"titles": [c.title for c in fresh]}}
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/{chat_id}/messages")
|
||||
async def post_message(
|
||||
request: Request,
|
||||
@@ -150,6 +167,7 @@ async def post_message(
|
||||
assistant_message = chat_service.create_message(
|
||||
db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=chat.model_id
|
||||
)
|
||||
generation_service.ensure(chat.id, assistant_message.id)
|
||||
|
||||
# `user` is required by the shared message template, which renders both
|
||||
# roles; without it the user bubble's initial blows up.
|
||||
@@ -187,7 +205,7 @@ async def stream_message(
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That message no longer exists.")
|
||||
|
||||
return StreamingResponse(
|
||||
_generate(chat.id, message.id),
|
||||
_follow(chat.id, message.id),
|
||||
media_type="text/event-stream",
|
||||
headers={
|
||||
"Cache-Control": "no-cache, no-transform",
|
||||
@@ -199,162 +217,48 @@ async def stream_message(
|
||||
)
|
||||
|
||||
|
||||
def _plain_text(content: str | list) -> str:
|
||||
"""The text of a message payload, whether it is a string or content parts."""
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
return " ".join(
|
||||
part.get("text", "")
|
||||
for part in content
|
||||
if isinstance(part, dict) and part.get("type") == "text"
|
||||
).strip()
|
||||
return ""
|
||||
async def _follow(chat_id: str, message_id: str) -> AsyncIterator[str]:
|
||||
"""Stream a generation that is running independently of this request.
|
||||
|
||||
This connection only *watches*. Closing it -- navigating away, opening
|
||||
another chat -- leaves the reply being written, and reconnecting replays
|
||||
the whole state immediately rather than starting over.
|
||||
|
||||
async def _generate(chat_id: str, message_id: str) -> AsyncIterator[str]:
|
||||
"""Drive one completion and frame it as SSE.
|
||||
|
||||
Opens its own database session rather than using the request's: streaming
|
||||
outlives the request handler, and the dependency-scoped session may already
|
||||
be closed by the time the first token arrives.
|
||||
Both `render` and `reasoning` carry the complete block each time rather
|
||||
than a delta, which is what makes reattaching mid-reply work at all: a
|
||||
follower arriving late has no earlier fragments to append to.
|
||||
"""
|
||||
accumulated: list[str] = []
|
||||
thinking: list[str] = []
|
||||
error: str | None = None
|
||||
reasoning_ms = 0
|
||||
generation = generation_service.ensure(chat_id, message_id)
|
||||
generation.followers += 1
|
||||
seen = -1
|
||||
|
||||
try:
|
||||
while True:
|
||||
if generation.version != seen:
|
||||
seen = generation.version
|
||||
if generation.thinking:
|
||||
yield sse.event("reasoning", escape_text(generation.thinking))
|
||||
if generation.content:
|
||||
yield sse.event("render", render_markdown(generation.text))
|
||||
|
||||
if generation.done:
|
||||
break
|
||||
# Polling rather than per-follower wakeups: the producer already
|
||||
# works in RENDER_INTERVAL steps, so a short sleep is simpler and
|
||||
# cannot drop a notification.
|
||||
await asyncio.sleep(generation_service.RENDER_INTERVAL * 0.8)
|
||||
finally:
|
||||
generation.followers = max(0, generation.followers - 1)
|
||||
|
||||
# The producer writes the message before marking itself done, so by here
|
||||
# the row is authoritative and the final bubble can be rendered from it.
|
||||
with session_scope() as db:
|
||||
chat = db.get(Chat, chat_id)
|
||||
message = db.get(Message, message_id)
|
||||
if chat is None or message is None:
|
||||
chat = db.get(Chat, chat_id)
|
||||
if message is None or chat is None:
|
||||
yield sse.event("close", "")
|
||||
return
|
||||
|
||||
first_user_text = ""
|
||||
# Handles models that emit <think> tags inline in content rather than
|
||||
# using the reasoning_content field.
|
||||
splitter = ReasoningSplitter()
|
||||
started = time.monotonic()
|
||||
reasoning_started: float | None = None
|
||||
last_render = 0.0
|
||||
dirty = False
|
||||
stopped = False
|
||||
|
||||
try:
|
||||
endpoint, model_id = chat_service.resolve_endpoint(db, chat)
|
||||
payload = chat_service.build_request(db, chat, upto=message)
|
||||
# A multimodal turn's content is a list of parts, not a string, so
|
||||
# the text has to be picked out before it can title a chat.
|
||||
first_user_text = next(
|
||||
(
|
||||
_plain_text(m["content"])
|
||||
for m in reversed(payload["messages"])
|
||||
if m["role"] == ROLE_USER
|
||||
),
|
||||
"",
|
||||
)
|
||||
|
||||
async for chunk in stream_chat(endpoint, payload):
|
||||
# A dedicated reasoning field is unambiguous; take it as-is.
|
||||
thought = delta_reasoning(chunk)
|
||||
if thought:
|
||||
if reasoning_started is None:
|
||||
reasoning_started = time.monotonic()
|
||||
thinking.append(thought)
|
||||
yield sse.event("reasoning", escape_text(thought))
|
||||
await asyncio.sleep(0)
|
||||
|
||||
text = delta_text(chunk)
|
||||
if not text:
|
||||
continue
|
||||
|
||||
for kind, piece in splitter.feed(text):
|
||||
if kind == REASONING:
|
||||
if reasoning_started is None:
|
||||
reasoning_started = time.monotonic()
|
||||
thinking.append(piece)
|
||||
yield sse.event("reasoning", escape_text(piece))
|
||||
else:
|
||||
# First answer token ends the thinking phase.
|
||||
if reasoning_started is not None and not reasoning_ms:
|
||||
reasoning_ms = int((time.monotonic() - reasoning_started) * 1000)
|
||||
accumulated.append(piece)
|
||||
dirty = True
|
||||
# Hand control back so the event is flushed rather than
|
||||
# batched behind a fast generator.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
# Re-render the answer so far, at most every RENDER_INTERVAL.
|
||||
# Markdown is rendered whole rather than appended, because a
|
||||
# list or a code fence is only correct once its context is
|
||||
# known -- and partial syntax resolves itself as more arrives.
|
||||
now = time.monotonic()
|
||||
if dirty and now - last_render >= RENDER_INTERVAL:
|
||||
yield sse.event("render", render_markdown("".join(accumulated)))
|
||||
last_render, dirty = now, False
|
||||
await asyncio.sleep(0)
|
||||
|
||||
if message_id in _CANCELLED:
|
||||
stopped = True
|
||||
break
|
||||
|
||||
for kind, piece in splitter.flush():
|
||||
if kind == REASONING:
|
||||
thinking.append(piece)
|
||||
yield sse.event("reasoning", escape_text(piece))
|
||||
else:
|
||||
accumulated.append(piece)
|
||||
|
||||
except LLMError as exc:
|
||||
error = exc.message
|
||||
log.info("generation failed for chat %s: %s", chat_id, exc.message)
|
||||
except asyncio.CancelledError:
|
||||
# The reader navigated away or closed the tab. Keep whatever was
|
||||
# produced so the partial reply is still there on reload.
|
||||
_CANCELLED.discard(message_id)
|
||||
message.content = "".join(accumulated)
|
||||
message.reasoning = "".join(thinking)
|
||||
message.complete = True
|
||||
message.stopped = True
|
||||
db.commit()
|
||||
raise
|
||||
except Exception as exc: # noqa: BLE001 - must not kill the stream silently
|
||||
error = "Something went wrong while generating this reply."
|
||||
log.exception("unexpected generation failure for chat %s: %s", chat_id, exc)
|
||||
|
||||
if reasoning_started is not None and not reasoning_ms:
|
||||
# Reasoning ran to the end without an answer following it.
|
||||
reasoning_ms = int((time.monotonic() - reasoning_started) * 1000)
|
||||
|
||||
_CANCELLED.discard(message_id)
|
||||
|
||||
message.content = "".join(accumulated)
|
||||
message.reasoning = "".join(thinking)
|
||||
message.reasoning_ms = reasoning_ms
|
||||
message.error = error or ""
|
||||
message.complete = True
|
||||
message.stopped = stopped
|
||||
log.debug(
|
||||
"chat %s: %d chars answer, %d chars reasoning, %.1fs total",
|
||||
chat_id,
|
||||
len(message.content),
|
||||
len(message.reasoning),
|
||||
time.monotonic() - started,
|
||||
)
|
||||
|
||||
if not chat.title_generated and (accumulated or error):
|
||||
chat.title = (
|
||||
await chat_service.generate_title(
|
||||
endpoint, model_id, first_user_text, message.content
|
||||
)
|
||||
if not error and first_user_text
|
||||
else chat_service.fallback_title(first_user_text)
|
||||
)
|
||||
chat.title_generated = True
|
||||
|
||||
db.commit()
|
||||
|
||||
final_html = templates.get_template("chat/_message.html").render(
|
||||
{
|
||||
"message": message,
|
||||
@@ -369,9 +273,7 @@ async def _generate(chat_id: str, message_id: str) -> AsyncIterator[str]:
|
||||
},
|
||||
}
|
||||
)
|
||||
title_html = templates.get_template("chat/_title_oob.html").render(
|
||||
{"chat": chat}
|
||||
)
|
||||
title_html = templates.get_template("chat/_title_oob.html").render({"chat": chat})
|
||||
|
||||
yield sse.event("done", final_html + title_html)
|
||||
yield sse.event("close", "")
|
||||
@@ -485,9 +387,10 @@ async def edit_message(
|
||||
db.delete(later)
|
||||
db.commit()
|
||||
|
||||
chat_service.create_message(
|
||||
assistant = chat_service.create_message(
|
||||
db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=chat.model_id
|
||||
)
|
||||
generation_service.ensure(chat.id, assistant.id)
|
||||
log.info("chat %s rewound to a message, %d discarded", chat.id, len(discarded))
|
||||
|
||||
return templates.TemplateResponse(
|
||||
@@ -507,7 +410,7 @@ async def stop_message(db: Db, user: RequiredUser, chat_id: str, message_id: str
|
||||
if message is None or message.chat_id != chat.id:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That message no longer exists.")
|
||||
|
||||
_CANCELLED.add(message.id)
|
||||
generation_service.request_stop(message.id)
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
@@ -648,6 +551,7 @@ async def regenerate(
|
||||
message.complete = False
|
||||
message.model_id = chat.model_id
|
||||
db.commit()
|
||||
generation_service.ensure(chat.id, message.id)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
|
||||
@@ -117,6 +117,12 @@ async def chat_detail(request: Request, db: Db, user: RequiredUser, chat_id: str
|
||||
if chat is None or chat.user_id != user.id:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That chat no longer exists.")
|
||||
|
||||
# Opening the chat is what "read" means.
|
||||
if chat.unread:
|
||||
chat.unread = False
|
||||
chat.unread_notified = False
|
||||
db.commit()
|
||||
|
||||
messages = list(
|
||||
db.scalars(
|
||||
select(Message).where(Message.chat_id == chat.id).order_by(Message.created_at)
|
||||
|
||||
Reference in New Issue
Block a user