Blame for backend/app/routers/support.py

"""Help bot (bottom right widget). Answers how to questions from the help corpus only, never from a writer's own
notebooks or posts (see support/scope.py). Streams over SSE.

POST /api/support/chat/stream          events: token, answer_done, done, error
GET  /api/support/conversations/latest restore the chat after a refresh
POST /api/support/escalate             "talk to a human" / feature request form, emailed to the team
"""

import json
import logging
import re
import time
import uuid
from collections import defaultdict, deque
from collections.abc import AsyncIterator

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy import select

from app.auth import Ctx, get_ctx
from app.config import settings
from app.llm.postprocess import strip_em_dashes
from app.models import SupportConversation, SupportMessage
from app.models.base import utcnow
from app.services.email import email_service
from app.services.plans import effective_plan
from app.support import escalation as esc
from app.support import llm, memory, prompts, scope
from app.support.llm import LLMError
from app.support.retriever import get_retriever

log = logging.getLogger("notestack.support")
router = APIRouter(prefix="/api/support", tags=["support"])

MAX_ESCALATIONS = 3
DOC_BUDGET_CHARS = 3500
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
_hits: dict[uuid.UUID, deque[float]] = defaultdict(deque)  # per process; enough to stop a runaway client


class ChatIn(BaseModel):
    message: str = Field(min_length=1, max_length=4000)
    page_path: str | None = Field(default=None, max_length=300)
    conversation_id: uuid.UUID | None = None


class EscalateIn(BaseModel):
    email: str = Field(min_length=3, max_length=320)
    concern: str = Field(min_length=1, max_length=4000)
    reason: str | None = Field(default=None, max_length=32)
    page_path: str | None = Field(default=None, max_length=300)
    conversation_id: uuid.UUID | None = None


def _rate_limit(user_id: uuid.UUID) -> None:
    now, window = time.monotonic(), _hits[user_id]
    while window and now - window[0] > 60:
        window.popleft()
    if len(window) >= settings.support_messages_per_minute:
        raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS,
                            "You are sending messages quickly. Try again in a minute.")
    window.append(now)


def _sse(event: str, data) -> str:
    return f"event: {event}\ndata: {json.dumps(data)}\n\n"


def _excerpt(body: str, query: str, budget: int = DOC_BUDGET_CHARS) -> str:
    """Whole doc when it fits, otherwise the sections that share the most words with the question."""
    if len(body) <= budget:
        return body
    words = set(re.findall(r"[a-z0-9]+", query.lower()))
    sections = re.split(r"(?m)^(?=#{1,3}\s)", body)
    ranked = sorted(range(len(sections)),
                    key=lambda i: -len(words & set(re.findall(r"[a-z0-9]+", sections[i].lower()))))
    keep, used = set(), 0
    for i in ranked:
        if used + len(sections[i]) > budget and keep:
            continue
        keep.add(i)
        used += len(sections[i])
    return "".join(sections[i] for i in sorted(keep))[:budget]


def _docs_block(scored, query: str) -> str:
    if not scored:
        return "(no matching documents)"
    return "\n\n".join(f"--- id: {s.doc.id}\ntitle: {s.doc.title}\n{_excerpt(s.doc.body, query)}" for s in scored)


def _owned(ctx: Ctx, conversation_id: uuid.UUID) -> SupportConversation:
    conv = ctx.db.get(SupportConversation, conversation_id)
    if not conv or conv.user_id != ctx.user.id:
        raise HTTPException(404, "Conversation not found")
    return conv


def _message_out(m: SupportMessage) -> dict:
    return {"id": str(m.id), "role": m.role, "content": m.content, "page_path": m.page_path,
            "cited_docs": m.cited_docs or [], "created_at": m.created_at.isoformat()}


@router.post("/chat/stream")
async def chat_stream(body: ChatIn, ctx: Ctx = Depends(get_ctx)) -> StreamingResponse:
    if not settings.support_enabled:
        raise HTTPException(404, "Help chat is off")
    _rate_limit(ctx.user.id)
    db, message = ctx.db, body.message.strip()
    page = scope.scrub_path(body.page_path)  # ids never reach the prompt, the database or the logs
    conv = _owned(ctx, body.conversation_id) if body.conversation_id else None
    if conv is None:
        conv = SupportConversation(user_id=ctx.user.id, title=message[:120], summary="", session_state={})
        db.add(conv)
        db.flush()
    conv_id = conv.id
    plan = effective_plan(db, ctx.workspace).id

    recent = memory.recent_messages(db, conv_id)
    question_reason = esc.classify_question(message)
    out_of_scope = scope.about_user_data(message)
    short_circuit = esc.should_short_circuit(question_reason) and not out_of_scope

    scored, messages = [], []
    if not out_of_scope and not short_circuit:
        scored = get_retriever().retrieve(
            message, history=memory.user_texts(recent), page_path=page, last_cited=memory.last_cited(recent)
        )
        system = prompts.ANSWER_PROMPT.format(
            docs=_docs_block(scored, message), user_context=memory.state_block(conv.session_state or {}),
            summary=conv.summary or "(none)",
        )
        messages = [{"role": "system", "content": system}]
        messages += [{"role": m.role, "content": m.content} for m in recent]
        messages.append({"role": "user", "content": f"[on {page}]\n{message}" if page else message})

    async def generate() -> AsyncIterator[str]:
        answer, escalate_reason, meta = "", None, llm.Meta()
        if out_of_scope:
            answer, broken = scope.out_of_scope_reply(message)
            escalate_reason = esc.Reason.HUMAN if broken else None
            yield _sse("token", answer)
        elif short_circuit:
            drafted = ""
            try:  # buffered, not streamed: the line is checked before the writer sees it
                async for tok in llm.stream_answer(
                    [{"role": "system", "content": esc.handoff_prompt(question_reason)},
                     {"role": "user", "content": message}]
                ):
                    drafted += tok
            except LLMError:
                drafted = ""
            drafted = strip_em_dashes(drafted.strip())
            answer = (drafted if drafted and esc.handoff_line_is_safe(drafted)
                      else esc.short_circuit_reply(question_reason, len(recent)))
            escalate_reason = question_reason
            yield _sse("token", answer)
        else:
            try:
                async for tok in llm.stream_answer(messages):
                    answer += tok
                    yield _sse("token", strip_em_dashes(tok))
            except LLMError:
                yield _sse("error", "The help assistant is unavailable right now. Please try again.")
                return
            answer = strip_em_dashes(answer.strip()) or "Sorry, I could not form an answer."
        yield _sse("answer_done", {"conversation_id": str(conv_id)})

        if not out_of_scope and not short_circuit:
            meta_system = prompts.META_PROMPT.format(
                doc_ids="\n".join(f"- {s.doc.id}: {s.doc.title}" for s in scored) or "(none)"
            )
            try:
                meta = await llm.complete_meta(
                    [{"role": "system", "content": meta_system}, {"role": "user", "content": message},
                     {"role": "assistant", "content": answer}]
                )
            except LLMError:
                meta = llm.Meta()  # the regex nets below still decide escalation
            escalate_reason = (
                (question_reason if question_reason is esc.Reason.FEATURE else None)
                or esc.classify_answer(answer)
                or (esc.Reason.HUMAN if meta.escalate else None)
            )

        valid = {s.doc.id for s in scored}
        citations = [c for c in meta.citations if c in valid]

        db.add(SupportMessage(conversation_id=conv_id, role="user", content=message, page_path=page))
        db.add(SupportMessage(conversation_id=conv_id, role="assistant", content=answer, page_path=page,
                              cited_docs=citations))
        conv.session_state = memory.update_state(conv.session_state or {}, page_path=page, cited=citations, plan=plan)
        db.commit()
        yield _sse("done", {"conversation_id": str(conv_id), "citations": citations,
                            "escalate": bool(escalate_reason),
                            "escalate_reason": escalate_reason.value if escalate_reason else None})

        # After `done`, so the extra call never delays the writer: fold older turns into the rolling summary.
        total = len(db.scalars(select(SupportMessage.id).where(SupportMessage.conversation_id == conv_id)).all())
        if total >= memory.SUMMARIZE_AT:
            fold = memory.to_fold(db, conv_id, recent)
            if fold:
                transcript = "\n".join(f"{m.role.upper()}: {m.content}" for m in fold)
                try:
                    conv.summary = strip_em_dashes(await llm.complete_text(
                        [{"role": "system", "content": prompts.SUMMARY_PROMPT},
                         {"role": "user", "content": f"Previous summary: {conv.summary or '(none)'}\n\n"
                                                      f"New turns:\n{transcript}"}],
                        max_tokens=500))[:1200]
                    db.commit()
                except LLMError:
                    log.info("summary update skipped")

    return StreamingResponse(generate(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@router.get("/conversations/latest")
def latest(ctx: Ctx = Depends(get_ctx)):
    conv = ctx.db.scalar(
        select(SupportConversation).where(SupportConversation.user_id == ctx.user.id)
        .order_by(SupportConversation.updated_at.desc()).limit(1)
    )
    if not conv:
        return {"conversation_id": None, "messages": []}
    msgs = ctx.db.scalars(
        select(SupportMessage).where(SupportMessage.conversation_id == conv.id)
        .order_by(SupportMessage.created_at).limit(60)
    ).all()
    return {"conversation_id": str(conv.id), "messages": [_message_out(m) for m in msgs]}


@router.post("/escalate", status_code=status.HTTP_202_ACCEPTED)
def escalate(body: EscalateIn, ctx: Ctx = Depends(get_ctx)):
    from_email = body.email.strip()
    if not _EMAIL.match(from_email):
        raise HTTPException(422, "Please enter a valid email address.")
    concern = body.concern.strip()
    if not concern:
        raise HTTPException(422, "Please describe what you need.")
    transcript: list[tuple[str, str]] = []
    if body.conversation_id:
        conv = _owned(ctx, body.conversation_id)
        if conv.escalation_count >= MAX_ESCALATIONS:
            raise HTTPException(429, "You have already sent this a few times. We will be in touch shortly.")
        rows = ctx.db.scalars(
            select(SupportMessage).where(SupportMessage.conversation_id == conv.id)
            .order_by(SupportMessage.created_at.desc()).limit(6)
        ).all()
        transcript = [(m.role, m.content[:500]) for m in reversed(rows)]
        conv.escalation_count += 1
        conv.updated_at = utcnow()
        ctx.db.commit()
    sent = email_service.send_support_escalation(
        from_email=from_email, name=ctx.user.name, plan=effective_plan(ctx.db, ctx.workspace).id,
        reason=body.reason or "human", concern=concern, page=scope.scrub_path(body.page_path), transcript=transcript,
    )
    if not sent:
        raise HTTPException(502, "We could not send that right now. Please try again.")
    return {"ok": True}